市面常见消息队列中间件对比
技术名称吞吐量 /IO/并发时效性（类似延迟）消息到达时间可用性可靠性优势应用场景activemq万级高高高简单易学中小型企业、项目rabbitmq万级极高（微秒）高极高生态好（基本什么语言都支持）、时效性高、易学适合绝大数的分布式应用kafka10万 QBS高（毫秒）极高极高吞吐量大、可靠性、可用性、强大的数据流处理能力适合大规模处理数据的场景、比如构建日志手机系统、实时数据传输、事件流收集传输rocketmq10万 QBS高ms极高极高吞吐量大、可靠性、可用性、可扩展性适用于金融等可靠性要求较高的场景、适合大规模的消息处理。金融、电商、大规模社交pulsar10万 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
入门
依赖引入
单消费者和生产者
一对一的关系
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");
//建立连接、创建频道
//频道，类似客户端，用于调用server
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//创建队列，队列持久化，第二份参数设置为 true
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(" [*] 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 参数：
queue：这是一个字符串参数，代表要消费的队列的名称。如果队列不存在，则会抛出异常。onMessage：这是一个回调函数，当有新的消息到达时会被调用。该函数需要接收两个参数：一个表示消息内容的Delivery对象和一个表示通道的Channel对象。consumerTag：这是一个可选参数，用于标识消费者。如果没有指定，则会自动生成一个唯一的标识符。autoAck：这是一个布尔值参数，表示是否自动确认消息。如果设置为true，则在消息被处理后会自动发送确认信息；如果设置为false，则需要手动发送确认信息。默认值为false。arguments：这是一个可选参数，用于设置消费者的其他属性，比如消息的最大长度、最大优先级等。
在使用channel.basicConsume时，需要注意以下几点：
队列名称：队列名称应该是唯一的，否则会抛出异常。消息处理：在onMessage回调函数中，需要对消息进行处理，并根据需要发送确认信息。消费者标识符：可以通过设置consumerTag来标识消费者，以便在后续操作中进行识别和管理。消费者属性：可以通过设置消费者的其他属性来控制消费者的行为，比如设置消息的最大长度、最大优先级等。
多消费者
多个消费者，比如一个工厂生产商品，一个商店卖不完，分给多个商店一起卖
生产者代码和上面一样
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 -> {
});
}
}
}