Youzi - RabbitMQ
type: channel
link: https://www.autisn.com/index.php/tag/RabbitMQ/
items: RabbitMQ知识点
items: Docker安装RabbitMQ并简单简单使用

RabbitMQ知识点
type: item
date: 2022-07-05T15:06:00+08:00
description: 1、消息应答的方法：Channel.basicAck(用于肯定确认)，RabbitMQ已知道该消息并且成功处理，可以将其丢弃Channel.basicNack(用于否定确认)Channel.basicReject(用于否定确认)，与Channel.basicNack相比少了一个参数，不处理该消息了，直接拒绝，可以将其丢弃了。2、Multiple的解释true表示批量应答channel上未应答的消息，比如channel上有传送tag的消息5,6,7,8,，当前tag是8，那么此时5-8的这些还未应答的消息就会被确认收到消息应答false同上面相比只会应答tag=8的消息，5,6,7这三个消息依然不会被确认收到消息应答手动应答的好处是可以批量应答并且减少网络拥堵 。3、消息手动应答代码//getEnvelope()属性，getDeliveryTag()标记 /* * 参数1：消息的标记 tag * 参数2：是否批量应答，false：不批量应答；true：批量 * */ channel.basicAck(message.getEnvelope().getDeliveryTag(), false); // 采用手动应答 boolean autoAck = false; channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,(consumerTag) -> { System.out.println(consumerTag+"消费者取消消费接口回调逻辑"); });4、持久化// 声明队列 // 持久化 需要让Queue持久化 boolean durable = true; channel.queueDeclare(TASK_QUEUE_NAME,durable,false,false,null);注意：如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新创建一个持久化的队列，不然就会出现错误。将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。5、不公平分发//不公平分发channel //basicQos(0) --> 轮询分发（公平分发），默认值为0 //channel.basicQos(1) --> 不公平分发，能者多劳 //channel.basicQos(>2) --> 预取值 int prefetchCount = 1; channel.basicQos(prefetchCount); //采用手动应答 boolean autoAck = false; channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);6、发布确认发布确认的策略:开启发布确认的方法：发布确认默认是没有开启的，如果要开启，需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在 channel 上调用该方法。//开启发布确认 channel.confirmSelect();6.1单个确认发布这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布，waitForConfirmsOrDie(long) 这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。这种确认方式有一个最大的缺点就是：发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。/** * 单个发送 */ public static void publishMessageIndividually() throws Exception { Channel channel = RabbitMqUtils.getChannel(); //队列声明 String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, true, false, false, null); //开启发布确认 channel.confirmSelect(); long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + "…

Docker安装RabbitMQ并简单简单使用
type: item
date: 2022-06-30T17:33:00+08:00
description: 1.Docker安装RabbitMQ1.1下载RabbitMQ镜像docker pull rabbitmq1.2创建并启动RabbitMQ容器docker run -id --hostname myrabbit --name rabbitmq -p 15672:15672 -p 5672:5672 rabbitmq1.3进入容器交互页面docker exec -it rabbitmq /bin/bash1.4下载插件rabbitmq-plugins enable rabbitmq_management1.5登录ip+端口号访问账号密码均为：guest1.6拓展# 创建账号和密码 rabbitmqctl add_user 用户名 密码 # 设置用户角色 rabbitmqctl set_user_tags admin administrator # 为用户添加资源权限 # set_permissions [-p <vhostpath>] <user> <conf> <write> <read> # 添加配置、写、读权限 rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"用户级别：administrator：可以登录控制台、查看所有信息、可以对 rabbitmq 进行管理。monitoring：监控者 登录控制台，查看所有信息。policymaker：策略制定者 登录控制台，指定策略。managment：普通管理员 登录控制台。注：有可能会出现下面这个问题访问Docker的rabbitMq容器的Web-UI界面无法查看消息的状态及出现Stats in management UI are disabled on this node解决办法：进入RabbitMQ容器docker exec -it rabbitmq /bin/bash进入容器后，cd到路径/etc/rabbitmq/conf.d/cd /etc/rabbitmq/conf.d/再修改management_agent.disable_metrics_collectorecho management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf退出容器exit输入指令docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_management重启RabbitMQdocker restart rabbitmq2.简单使用2.1pom.xml文件添加依赖(本环境是jdk1.8的) com.rabbitmq amqp-client 5.8.0 commons-io commons-io 2.6 2.2新建Producer类为生产者package com.youzi.rabbitmq.one; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * @Author youzi * @ClassName Producer * @Date 2022/6/30 15:41 * @Description //TODO 生产者 */ public class Producer { //设置队列名称 public static final String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { //1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //2.设置IP factory.setHost("ip地址"); //3.设置用户名密码 factory.setUsername("用户名"); factory.setPassword("密码"); //4.创建连接 Connection connection = factory.newConnection(); //5.获取信道 Channel channel = connection.createChannel(); /** * …