RabbitMQ入门实战

官方：https://www.rabbitmq.com

基本概念

AMQP协议：https://www.rabbitmq.com/tutorials/amqp-concepts.html
定义：高级信息队列协议（Advanced Message Queue Protocol）
生产者：发消息到某个交换机
消费者：从某个队列中取信息
交换机（Exchange）:负责把消息转发到对应的队列
队列（Queue）：存储消息
路由（Routes）:将消息从一个地方转发到另一个地方
AMQP模型：

安装

windows 安装：https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.12.0
安装 erlang 25.3.2（因为 RabbitMQ 依赖 erlang,不安装这个安装RabbitMQ会报错），这个语言的性能非常高。
erlang 下载：https://www.erlang.org/patches/otp-25.3.2

安装完 erlang 后，安装 rabbitmq 即可。
win + r 打开 services.msc（服务菜单），查看 rabbitmq 服务是否已启动：

安装 rabbitmq 监控面板：
在 rabbitmq 安装目录的 sbin 中执行下述脚本：
D:\software\rabbitmq\rabbitmq_server-3.12.0\sbin

rabbitmq-plugins.bat enable rabbitmq_management

访问：http://localhost:15672，用户名密码都是 guest：

如果想要在远程服务器安装访问 rabbitmq 管理面板，你要自己创建一个管理员账号，不能用默认的 guest，否则会被拦截（官方出于安全考虑）。

如果被拦截，可以自己创建管理员用户：
参考文档的 Adding a User：https://www.rabbitmq.com/access-control.html

rabbitmq 端口占用：
5672：程序连接的端口
15672：webUI

快速入门

单向发送

一个生产者给一个队列发消息，一个消费者从这个队列取消息，一对一

引入消息队列 Java 客户端：

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

生产者代码：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.nio.charset.StandardCharsets;public class Send {private final static String QUEUE_NAME = "hello";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(QUEUE_NAME, false, false, false, null);String message = "Hello World!";channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));System.out.println(" [x] Sent '" + message + "'");}}
}

Channel频道：理解为操作消息队列的client，提供了和消息队列server建立通信的方法（为了复用连接，提高传输效率）。程序通过channel操作rabbitmq
创建消息队列：
参数：
queueName:消息队列名称（注意，同名称的消息队列，只能用同样的参数创建一次）
durabale：消息队列重启后，消息是否丢失
exclusive：是否只允许当前这个创建消息队列的连接操作消息队列
autoDelete：没有人用队列后，是否要删除队列
执行程序后，可以看到有 1 条消息：

消费者代码：

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);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");// 定义了如何处理消息DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), StandardCharsets.UTF_8);System.out.println(" [x] Received '" + message + "'");};// 消费消息，会持续阻塞channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });}
}

启动消费者后，可以看到消息被消费了

多消费者

官方教程：https://www.rabbitmq.com/tutorials/tutorial-two-java.html
场景：多个机器同时去接受并处理任务（尤其是每个机器的处理能力有限）
一个生产者给一个队列发消息，多个消费者从这个队列中取消息

1）队列持久化
durable参数设置为true,服务器重启后队列不丢失：

channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

2）消息持久化
指定MessageProperties.PERSISTENT_TEXT_PLAIN 参数：

channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));

生产者代码：

package com.yupi.springbootinit.mq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;import java.util.Scanner;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();//在指定的交换机上发布消息到TASK_QUEUE_NAME的队列中，使用 MessageProperties.PERSISTENT_TEXT_PLAIN将消息标为持久化，最后将消息体转换为字节数组channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}}}

消费者代码：

package com.yupi.springbootinit.mq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;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");final Connection connection = factory.newConnection();for (int i = 0; i < 2; i++) {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 + "'");// 停 20 秒，模拟机器处理能力有限Thread.sleep(20000);} catch (InterruptedException e) {e.printStackTrace();} finally {System.out.println(" [x] Done");// 手动发送应答,告诉RabbitMQ消息已经被处理channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}};// 开始消费消息,传入队列名称,是否自动确认,投递回调和消费者取消回调channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {});}}
}

这样你就可以跑通你的第一个RabbitMQ并且了解了单向发送和多消费者两种方式，下期分享RabbitMQ一个重要的概念—交换机