文章目录
- RabbitMQ入门实战
- 基本概念
- 安装
- 快速入门
- 单向发送
- 多消费者
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一个重要的概念—交换机
