RabbitMQ 消息队列安装及入门

市面常见消息队列中间件对比

技术名称	吞吐量 /IO/并发	时效性（类似延迟）消息到达时间	可用性	可靠性	优势	应用场景
activemq	万级	高	高	高	简单易学	中小型企业、项目
rabbitmq	万级	极高（微秒）	高	极高	生态好（基本什么语言都支持）、时效性高、易学	适合绝大数的分布式应用
kafka	10万 QBS	高（毫秒）	极高	极高	吞吐量大、可靠性、可用性、强大的数据流处理能力	适合大规模处理数据的场景、比如构建日志手机系统、实时数据传输、事件流收集传输
rocketmq	10万 QBS	高ms	极高	极高	吞吐量大、可靠性、可用性、可扩展性	适用于金融等可靠性要求较高的场景、适合大规模的消息处理。金融、电商、大规模社交
pulsar	10万 QBS	高ms	高	极高	可靠性、可用性很高、新兴（技术架构先进）	适合大规模、高并发的分布式系统（云原生）适合实时分析、事件流处理、物联网数据处理。

RabbitMQ 

RabbitMQ 是基于 AMQP 高级消息队列协议的。

 实际使用可根据官方文档的 demo 。

官方文档：RabbitMQ Tutorials | RabbitMQ

模型

生产者：通俗就是发消息的人，比如在外卖软件上点餐的人

消费者：通俗就是处理消息的任务，比如外卖软件上的商家，需要根据顾客的要求制作餐

交换机：负责把消息转发到对应的队列中，比如有外卖的时候，系统给附近的外面小哥派单

队列：存放消息的地方，等待消费者消费，比如商家肯定不是只做一份餐，做好的餐放在一个指定的位置等待外卖小哥来取餐

路由：转发，就是怎么把消息从一个地方转到另一个地方，通常加在交换机和队列之间，比如系统指定某个范围的外卖小哥接这单

安装

1. 首先安装 RabbitMQ，直接官网下载即可，如果下载速度慢，可以换个网络，或者找找有没有国内镜像。（当初我下载的时候找了半天的镜像都是版本比较老的，结果想着挂一晚上下载，结果官网突然就快了，白折腾了。）

官方网站：Installing on Windows | RabbitMQ

国内镜像：Index of rabbitmq-server-local/v3.12.7

一路 next ，傻瓜式安装即可

安装之后检查服务中是否已经运行了。

2. 安装监控面板

在 RabbitMQ 安装目录下的 sbin 目录下的CMD 输入下面的命令

rabbitmq-plugins.bat enable rabbitmq_management

 安装成功：

默认端口号是 5672，webUI 是 15672

在浏览器输入地址打开管理界面：http://localhost:15672

默认账号密码是 guest

注意：1. 安装目录不能是中文，不能有空格等非法字符，否则页面打不开

           2. 如果想要在远程服务器访问 RabbitMQ 管理面板，需要创建管理员账号，比如在宝塔面板使用时宝塔面板提供的 admin账号，地址就是宝塔面板的 IP 

创建账号：access-control | RabbitMQ

入门

依赖引入

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.20.0</version>
</dependency>

单消费者和生产者

一对一的关系

1. 生产者代码

public class SingleProducer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {//创建连接ConnectionFactory factory = new ConnectionFactory();//设置了本地连接，如果修改了用户名和密码，需要设置/*factory.setPassword();factory.setUsername();*/factory.setHost("localhost");//建立连接、创建频道//频道，类似客户端，用于调用serverConnection connection = factory.newConnection();Channel channel = connection.createChannel();//创建队列，队列持久化，第二份参数设置为 truechannel.queueDeclare(QUEUE_NAME, false, false, false, null);String message = "Hello World!";//发送消息channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));System.out.println(" [*] Waiting for messages. To exit press CTRL+C");}
}

channel 频道：理解为操作消息队列的 Client，通过 channel 收发消息，提供了和消息对了 server 建立通信的传输方法

channel.queueDeclare 方法参数：

queue：这是一个字符串参数，代表要声明的队列的名称。如果队列不存在，则会自动创建一个新的队列。

durable：这是一个布尔值参数，表示队列是否持久化。如果设置为true，则队列会在服务器重启后仍然存在；如果设置为false，则队列在服务器重启后会被删除。默认值为false。

exclusive：这也是一个布尔值参数，表示队列是否为独占模式。如果设置为true，则只有当前连接可以访问该队列；如果设置为false，则其他连接也可以访问该队列。默认值为false。

autoDelete：这是另一个布尔值参数，表示队列是否自动删除。如果设置为true，则当最后一个消费者取消订阅时，队列将被删除；如果设置为false，则队列将一直存在，直到手动删除或服务器重启。默认值为false。

arguments：这是一个可选参数，用于设置队列的其他属性，比如消息的最大长度、最大优先级等。

channel.basicPublish 参数：

exchange：这是一个字符串参数，代表交换机的名称。如果不需要使用特定的交换机，可以传递一个空字符串（""）。交换机是RabbitMQ中用于接收生产者发送的消息并根据绑定规则路由到队列的组件。

routingKey：这也是一个字符串参数，它指定了发布消息的队列。无论通道绑定到哪个队列，最终发布的消息都会包含这个指定的路由键。路由键是用来确定消息应该发送到哪个队列的重要信息。

message：这是要发布的消息本身，通常是字节数组的形式。

properties：这是一个可选参数，用于设置消息的属性，比如消息的优先级、过期时间等。

在使用channel.basicPublish时，需要注意以下几点：

exchange和routingKey不能为空：在AMQImpl类中的实现要求这两个参数都不能为null，否则会抛出异常。

交换机类型：根据不同的需求，可以选择不同类型的交换机，如fanout、direct或topic。每种类型的交换机都有其特定的路由规则。

非命名队列：在某些情况下，比如日志系统，可以使用非命名队列，这样消费者可以接收到所有相关的日志消息，而不是特定的部分。

2. 消费者代码

public class SingleConsumer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//声明队列，同一个消息队列参数必须一致channel.queueDeclare(QUEUE_NAME, false, false, false, null);//定义了如何处理消息DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};//接收、消费消息channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });System.out.println(" [*] Waiting for messages. To exit press CTRL+C");}
}

 channel.basicConsume 参数：

1. queue：这是一个字符串参数，代表要消费的队列的名称。如果队列不存在，则会抛出异常。
2. onMessage：这是一个回调函数，当有新的消息到达时会被调用。该函数需要接收两个参数：一个表示消息内容的Delivery对象和一个表示通道的Channel对象。
3. consumerTag：这是一个可选参数，用于标识消费者。如果没有指定，则会自动生成一个唯一的标识符。
4. autoAck：这是一个布尔值参数，表示是否自动确认消息。如果设置为true，则在消息被处理后会自动发送确认信息；如果设置为false，则需要手动发送确认信息。默认值为false。
5. arguments：这是一个可选参数，用于设置消费者的其他属性，比如消息的最大长度、最大优先级等。

在使用channel.basicConsume时，需要注意以下几点：

1. 队列名称：队列名称应该是唯一的，否则会抛出异常。
2. 消息处理：在onMessage回调函数中，需要对消息进行处理，并根据需要发送确认信息。
3. 消费者标识符：可以通过设置consumerTag来标识消费者，以便在后续操作中进行识别和管理。
4. 消费者属性：可以通过设置消费者的其他属性来控制消费者的行为，比如设置消息的最大长度、最大优先级等。

多消费者

多个消费者，比如一个工厂生产商品，一个商店卖不完，分给多个商店一起卖

生产者代码和上面一样

public class MultiProducer {private static final String TASK_QUEUE_NAME = "multi_queue";public static void main(String[] argv) throws Exception {//创建连接ConnectionFactory factory = new ConnectionFactory();//设置本地连接factory.setHost("localhost");//创建队列，创建频道，类似客户端try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {//队列持久化channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);//设置消息Scanner scanner = new Scanner(System.in);while(scanner.hasNext()){//输入消息String message = scanner.nextLine();//发送消息channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}}
}

控制处理任务的积压数，最多同时处理任务数

channel.basicQos(1); //最多处理1个

消息确认机制

ack 确认、nack 消息失败、reject 拒绝

当消息拿走之后会有一个确认机制，保证消息成功被消费。当消费者接收消息会给一个反馈，确认消息的状态，成功消息才会被移除。

支持配置 autoack ，建议修改为 false，根据实际情况手动确认。

//手动确认
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);

//手动拒绝
channel.basicNack(delivery.getEnvelope().getDeliveryTag(),false,false);

 消费者代码

public class MultiConsumer {private static final String TASK_QUEUE_NAME = "multi_queue";public static void main(String[] argv) throws Exception {//创建连接ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");for (int i = 0; i < 2; i++) {final Connection connection = factory.newConnection();final Channel channel = connection.createChannel();//队列持久化，参数要一致channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");//控制处理任务的积压数，最多同时处理任务数channel.basicQos(1);//定义了如何处理消息int finalI = i;DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");try {//处理工作的逻辑System.out.println(" [x] Received '" +"消费者：" + finalI + " 消息："+ message + "'");//睡一定时间，模拟机器处理能力有限Thread.sleep(20000);channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {System.out.println(" [x] Done");channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}};//接收消息，消费消息，开启消息监听channel.basicConsume(TASK_QUEUE_NAME, false , deliverCallback, consumerTag -> {});}}
}