为什么使用消息队列？
以用户下单购买商品的行为举例，在使用微服务架构时，我们需要调用多个服务，传统的调用方式是同步调用，这会存在一定的性能问题
使用消息队列可以实现异步的通信方式，相比于同步的通信方式，异步的方式可以让上游快速成功，极大提高系统的吞吐量
消息队列的使用场景有如下：
- 异步处理：以上述用户下单购买商品为例，将多个不关联的任务放进消息队列，提高系统性能
- 应用解耦：以上述用户下单购买商品为例，订单系统通知库存系统减库存，传统的做法是订单系统调用库存系统的接口，订单系统和库存系统高耦合，当库存系统出现故障时，订单就会失败。使用消息队列，用户下单后，订单系统完成持久化，将消息写入消息队列，返回用户下单成功。库存系统订阅下单消息，获取下单消息，进行减库存操作。就算库存系统出现故障，消息队列也能保证消息的可靠投递，不会导致系统异常
- 流量削峰：举行秒杀活动时，为防止流量过大导致应用挂掉，服务器收到用户请求后，先写入消息队列，如果超过了消息队列长度的最大值，则直接抛弃或跳转到错误页面。秒杀业务根据消息队列中的请求信息，再做后续处理，缓解短时间的流量高峰
RabbitMQ 安装
以 ubuntu 22.04.3 为例，参考 RabbitMQ 官网提供的安装脚本
#!/bin/sh
## 要想安装最新版本的 rabbitmq，可以选择 Cloudsmith 存储库下载，为此我们必须安装 apt-transport https 包
sudo apt-get install curl gnupg apt-transport-https -y
## 获取 Cloudsmith 存储库提供的签名密钥并添加到系统中，这样这样才能使用 Cloudsmith 仓库下载包
curl -1sLf "https://keys.openpgp.org/vks/v1/by-fingerprint/0A9AF2115F4687BD29803A206B73A36E6026DFCA" | sudo gpg --dearmor | sudo tee /usr/share/keyrings/com.rabbitmq.team.gpg > /dev/null
curl -1sLf https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.E495BB49CC4BBE5B.key | sudo gpg --dearmor | sudo tee /usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg > /dev/null
curl -1sLf https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9F4587F226208342.key | sudo gpg --dearmor | sudo tee /usr/share/keyrings/rabbitmq.9F4587F226208342.gpg > /dev/null
## 将描述 RabbitMQ 和 Erlang 包存储库的文件放在 /etc/apt/sources.list.d/ 目录下
sudo tee /etc/apt/sources.list.d/rabbitmq.list <<EOF
## Provides modern Erlang/OTP releases
##
deb [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa1.novemberain.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa1.novemberain.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main
# another mirror for redundancy
deb [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa2.novemberain.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa2.novemberain.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main
## Provides RabbitMQ
##
deb [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa1.novemberain.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa1.novemberain.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main
# another mirror for redundancy
deb [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa2.novemberain.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa2.novemberain.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main
EOF
## 更新存储库索引
sudo apt-get update -y
## 安装 erlang
sudo apt-get install -y erlang-base \
erlang-asn1 erlang-crypto erlang-eldap erlang-ftp erlang-inets \
erlang-mnesia erlang-os-mon erlang-parsetools erlang-public-key \
erlang-runtime-tools erlang-snmp erlang-ssl \
erlang-syntax-tools erlang-tftp erlang-tools erlang-xmerl
## 安装 rabbitmq
sudo apt-get install rabbitmq-server -y --fix-missing
使用以下命令启动、关闭和查看 rabbitmq 状态
sudo systemctl stop rabbitmq-server
sudo systemctl start rabbitmq-server
sudo systemctl status rabbitmq-server
要想访问 rabbitmq 的 web 管理界面，需要执行以下命令，启动 rabbitmq 的插件管理
rabbitmq-plugins enable rabbitmq_management
访问：http://127.0.0.1:15672/ 可查看 rabbitmq 的 web 管理界面，但首先要创建用户，这里创建一个管理员用户，使用该用户登录
# rabbitmqctl add_user {username} {password}
# 设置账号和密码
rabbitmqctl add_user admin 123
# rabbitmqctl set_user_tags {username} {role}
# 设置角色，administrator 是管理员角色
rabbitmqctl set_user_tags admin
# rabbitmqctl set_permissions [-p vhost] {user} {conf} {write} {read}
# 设置权限：
# {vhost} 表示待授权用户访问的 vhost 名称，默认为 "/"
# {user} 表示待授权访问特定 vhost 的用户名称
# {conf} 表示待授权用户的配置权限，是一个匹配资源名称的正则表达式
# {write} 表示待授权用户的写权限，是一个匹配资源名称的正则表达式
# {read} 表示待授权用户的读权限，是一个资源名称的正则表达式
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
更多 rabbitmqctl 命令可参考官网：https://www.rabbitmq.com/rabbitmqctl.8.html
RabbitMQ 组成与消息模型
RabbitMQ 是使用 Erlang 语言开发的开源的消息队列，基于 AMQP（高级消息队列协议）实现
RabbitMQ 的组成部分如下：
- Message：消息，又可分为消息头和消息体，消息头由一系列可选属性组成
- Producer：消息生产者，是向交换器发布消息的客户端应用程序
- Consumer：消息消费者，从消息队列获取消息的客户端应用程序
- Exchange：交换器，接收生产者发送的消息并路由到相应的队列，常用的交换器类型有 direct、fanout、topic
- Binding：绑定，用于关联消息队列和交换器
- Queue：消息队列，保存消息直到放送给消费者
- Rounting-key：路由键，决定信息该投递到哪个队列的规则
- Connection：链接，指应用与 rabbit 服务器建立的 TCP 链接
- Channel：信道，TCP 里面的虚拟链接，一条 TCP 链接上可以创建多条信道，可以避免频繁创建和销毁 TCP 连接所带来的开销
- Virtual Host：虚拟主机，表示一批交换器、消息队列和相关对象，是共享相同身份认证和加密环境的独立服务器域，可以理解为一个 mini 版的 RabbitMQ 服务器
- Broker：消息队列服务器实体
组件协同工作的执行流程如下：
- 消息生产者连接到 RabbitMQ Broker，创建 connection，开启 channel
- 生产者声明交换机类型、名称、是否持久化等
- 生产者发送消息，并指定消息是否持久化等属性，指定 routing key
- exchange 收到消息后，根据 routing key 将消息路由到跟当前交换机绑定的相匹配的队列
- 消费者监听消息队列，接收到消息后开始业务处理
从上述流程我们可以看到，消息首先要经过 Exchange 路由才能找到对应的 Queue。Exchange 有四种类型，分别是 Direct、Fanout、Topic、Headers
1. Direct Exchange
直连交换机，是一种带路由功能的交换机，需要绑定一个队列，绑定时要指定一个 RoutingKey（路由键）。生产者把消息发送到交换机时，也必须指定消息的 RoutingKey（路由键）。 Exchange 根据消息的 RoutingKey 进行判断，只有队列的 RoutingKey 与消息的 RoutingKey 一致，才会接收到消息
2. Fanout Exchange
扇形交换机，可以有多个消费者，每个消费者有自己的队列，每个队列都要绑定到交换机。生产者把消息发送到交换机，交换机把消息发给绑定过的所有队列，队列的消费者都能拿到消息，实现一条消息被多个消费者消费
3. Topic Exchange
主题交换机和直连交换机类似，也可以根据 RoutingKey 把消息路由到不同的队列，只不过 Topic 类型的交换机可以让队列在绑定 RoutingKey 时使用通配符。这种模型的 RoutingKey 一般由一个或多个单词组成，多个单词以 . 符号分割
通配符有两种：
*符号：匹配一个词，比如a.*可以匹配a.b、a.c，但是匹配不了a.b.c#符号：匹配一个或多个词，比如rabbit.#可以匹配rabbit.a.b、rabbit.a，也可以匹配rabbit.a.b.c
4. Headers Exchange
头部交换机不是用 RoutingKey 进行路由匹配，而是匹配请求头中所带的键值进行路由。创建队列需要设置绑定的头部信息，有两种模式：全部匹配和部分匹配，交换机会根据生产者发送过来的头部信息携带的键值去匹配队列绑定的键值，路由到对应的队列
SpringBoot 整合 RabbitMQ
生产者端引入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
生产者端添加配置
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: root
password: 123
生产者端配置队列、交换机
@Configuration
public class RabbitMqConfig {
@Bean
public Queue rabbitmqTestDirectQueue() {
// Direct 队列
// name：队列名称
// durable：是否持久化
// exclusive：是否独享，如果设置 true，则只有创建者可以使用此队列
// autoDelete: 是否自动删除，也就是临时队列，当最后一个消费者断开连接就会自动删除
return new Queue("test_direct_queue", true, false, false);
}
@Bean
public DirectExchange rabbitmqTestDirectExchange() {
// Direct 交换机
return new DirectExchange("test_direct_exchange", true, false);
}
@Bean
public Binding bindDirect() {
return BindingBuilder
// // 绑定 Direct 队列
.bind(rabbitmqTestDirectQueue())
//到 Direct 交换机
.to(rabbitmqTestDirectExchange())
// 并设置匹配键
.with("test_direct_routing");
}
@Bean
public Queue rabbitmqTestFanoutQueueA() {
// fanout 队列 a
return new Queue("test_fanout_queue_a", true, false, false);
}
@Bean
public Queue rabbitmqTestFanoutQueueB() {
// fanout 队列 b
return new Queue("test_fanout_queue_b", true, false, false);
}
@Bean
public FanoutExchange rabbitmqTestFanoutExchange() {
// Fanout 交换机
return new FanoutExchange("test_fanout_exchange", true, false);
}
@Bean
public Binding bindFanoutA() {
return BindingBuilder
// 绑定 Fanout 队列 a
.bind(rabbitmqTestFanoutQueueA())
//到 Fanout 交换机
.to(rabbitmqTestFanoutExchange());
}
@Bean
public Binding bindFanoutB() {
return BindingBuilder
// 绑定 Fanout 队列 b
.bind(rabbitmqTestFanoutQueueB())
//到 Fanout 交换机
.to(rabbitmqTestFanoutExchange());
}
@Bean
public Queue rabbitmqTestTopicQueueA() {
// topic 队列 a
return new Queue("test_topic_queue_a", true, false, false);
}
@Bean
public Queue rabbitmqTestTopicQueueB() {
// topic 队列 b
return new Queue("test_topic_queue_b", true, false, false);
}
@Bean
public TopicExchange rabbitmqTestTopicExchange() {
// Topic 交换机
return new TopicExchange("test_topic_exchange", true, false);
}
@Bean
public Binding bindTopicA() {
return BindingBuilder
// 绑定 Topic 队列 a
.bind(rabbitmqTestTopicQueueA())
//到 Topic 交换机
.to(rabbitmqTestTopicExchange())
// 并设置匹配键
.with("a.*");
}
@Bean
public Binding bindTopicB() {
return BindingBuilder
// 绑定 Topic 队列 b
.bind(rabbitmqTestTopicQueueB())
//到 Topic 交换机
.to(rabbitmqTestTopicExchange())
// 并设置匹配键
.with("b.*");
}
@Bean
public Queue rabbitmqTestHeadersQueueA() {
// headers 队列 a
return new Queue("test_headers_queue_a", true, false, false);
}
@Bean
public Queue rabbitmqTestHeadersQueueB() {
// headers 队列 b
return new Queue("test_headers_queue_b", true, false, false);
}
@Bean
public HeadersExchange rabbitmqTestHeadersExchange() {
// Headers 交换机
return new HeadersExchange("test_headers_exchange", true, false);
}
@Bean
public Binding bindHeadersA() {
Map<String, Object> map = new HashMap<>();
map.put("key_a1", "a1");
map.put("key_a2", "a2");
return BindingBuilder
// 绑定 Headers 队列 a
.bind(rabbitmqTestHeadersQueueA())
//到 Headers 交换机
.to(rabbitmqTestHeadersExchange())
// 全部匹配
.whereAll(map).match();
}
@Bean
public Binding bindHeadersB() {
Map<String, Object> map = new HashMap<>();
map.put("key_b1", "b1");
map.put("key_b2", "b2");
return BindingBuilder
// 绑定 Headers 队列 b
.bind(rabbitmqTestHeadersQueueB())
//到 Headers 交换机
.to(rabbitmqTestHeadersExchange())
// 部分匹配
.whereAny(map).match();
}
}
生产者端发送消息
@Controller
public class RabbitMQController {
@Autowired
private RabbitTemplate rabbitTemplate;
@GetMapping("sendDirectMsg")
public String sendDirectMsg() {
Map<String, Object> map = new HashMap<>();
map.put("msg", "test_send_direct_msg");
rabbitTemplate.convertAndSend("test_direct_exchange", "test_direct_routing", map);
return "OK";
}
@GetMapping("sendFanoutMsg")
public String sendFanoutMsg() {
Map<String, Object> map = new HashMap<>();
map.put("msg", "test_send_fanout_msg");
rabbitTemplate.convertAndSend("test_fanout_exchange", "", map);
return "OK";
}
@GetMapping("sendTopicMsgA")
public String sendTopicMsgA() {
Map<String, Object> map = new HashMap<>();
map.put("msg", "test_send_topic_msg_a");
rabbitTemplate.convertAndSend("test_topic_exchange", "a.c", map);
return "OK";
}
@GetMapping("sendTopicMsgB")
public String sendTopicMsgB() {
Map<String, Object> map = new HashMap<>();
map.put("msg", "test_send_topic_msg_b");
rabbitTemplate.convertAndSend("test_topic_exchange", "b.c", map);
return "OK";
}
@GetMapping("sendHeadersMsgA")
public String sendHeadersMsgA() {
MessageProperties msgp = new MessageProperties();
msgp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
msgp.setContentType("UTF-8");
Map<String, Object> map = new HashMap<>();
map.put("key_a1", "a1");
map.put("key_a2", "a2");
msgp.getHeaders().putAll(map);
Message msg = new Message("test_send_headers_msg_a".getBytes(), msgp);
rabbitTemplate.convertAndSend("test_headers_exchange", null, msg);
return "OK";
}
@GetMapping("sendHeadersMsgB")
public String sendHeadersMsgB() {
MessageProperties msgp = new MessageProperties();
msgp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
msgp.setContentType("UTF-8");
Map<String, Object> map = new HashMap<>();
map.put("key_b1", "a1");
map.put("key_b3", "b3");
msgp.getHeaders().putAll(map);
Message msg = new Message("test_send_headers_msg_b".getBytes(), msgp);
rabbitTemplate.convertAndSend("test_headers_exchange", null, msg);
return "OK";
}
}
消费端引入的依赖和配置与生产端一样，接收消息
/**
* Direct 队列消费者
*/
@Slf4j
@Component
@RabbitListener(queues = {"test_direct_queue"})
public class RabbitMqDirectReceiver {
@RabbitHandler
public void process(Map<String, Object> map) {
log.info("消费者接收到消息: {}", map.toString());
}
}
/**
* Fanout 队列消费者 a
*/
@Slf4j
@Component
@RabbitListener(queues = {"test_fanout_queue_a"})
public class RabbitMqFanoutAReceiver {
@RabbitHandler
public void process(Map<String, Object> map) {
log.info("消费者接收到消息: {}", map.toString());
}
}
/**
* Fanout 队列消费者 b
*/
@Slf4j
@Component
@RabbitListener(queues = {"test_fanout_queue_b"})
public class RabbitMqFanoutBReceiver {
@RabbitHandler
public void process(Map<String, Object> map) {
log.info("消费者接收到消息: {}", map.toString());
}
}
/**
* Topic 队列消费者 a
*/
@Slf4j
@Component
@RabbitListener(queues = {"test_topic_queue_a"})
public class RabbitMqTopicAReceiver {
@RabbitHandler
public void process(Map<String, Object> map) {
log.info("消费者接收到消息: {}", map.toString());
}
}
/**
* Topic 队列消费者 b
*/
@Slf4j
@Component
@RabbitListener(queues = {"test_topic_queue_b"})
public class RabbitMqTopicBReceiver {
@RabbitHandler
public void process(Map<String, Object> map) {
log.info("消费者接收到消息: {}", map.toString());
}
}
/**
* Topic 队列消费者 a
*/
@Slf4j
@Component
@RabbitListener(queues = {"test_topic_queue_a"})
public class RabbitMqTopicAReceiver {
@RabbitHandler
public void process(Map<String, Object> map) {
log.info("消费者接收到消息: {}", map.toString());
}
}
/**
* Headers 队列消费者 a
*/
@Slf4j
@Component
public class RabbitMqHeadersAReceiver {
@RabbitListener(queuesToDeclare = @Queue("test_headers_queue_a"))
public void process(Message msg) throws Exception {
MessageProperties msgp = message.getMessageProperties();
String contentType = msgp.getContentType();
log.info("消费者接收到消息: {}", new String(message.getBody(), contentType));
}
}
/**
* Headers 队列消费者 b
*/
@Slf4j
@Component
public class RabbitMqHeadersBReceiver {
@RabbitListener(queuesToDeclare = @Queue("test_headers_queue_b"))
public void process(Message msg) throws Exception {
MessageProperties msgp = message.getMessageProperties();
String contentType = msgp.getContentType();
log.info("消费者接收到消息: {}", new String(message.getBody(), contentType));
}
}
RabbitMQ 消息确认机制
消费者从队列中获取到消息之后，在处理消息时出现异常，那这条正在处理的消息就没有完成消息消费，数据就会丢失。生产者同样如此，生产者发消息给交换机，也不能保证消息准确发送过去了
RabbitMQ 的消息确认分为两部分：
- 消息发送确认：用来确认生产者是否成功将消息投递到队列
- 消息接收确认：用来确认消费者是否成功接收到消息
1. 生产端确认
RabbitMQ 提供了两种机制，用于告知生产端是否发送消息成功：
- publisher-confirm：消息投递交换机，返回成功/失败信息
- publisher-return：消息投递交换机成功，但路由到队列失败，返回失败信息
生产端配置如下：
spring:
rabbitmq:
# 开启 publisher-confirm
publisher-confirm-type: correlated
# 开启 publisher-return
publisher-returns: true
publish-confirm-type 有三个值：
- none：禁用发布确认模式，默认值
- simple：同步等待 confirm 结果，直到超时
- correlated：异步通知 confirm 结果，需要定义回调函数 ConfirmCallback
生产端配置 ConfirmCallback 函数和 ReturnCallback 函数
@Slf4j
@Configuration
public class ProviderCallBackConfig {
@Autowired
private CachingConnectionFactory factory;
@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate rabbitTemplate = new RabbitTemplate(factory);
// mandatory 设置为 true，若 exchange 无法找到合适的 queue 存储消息就会调用 basic.return 方法将消息返还给生产者
// mandatory 设置为 false 时，出现上述情况 broker 会直接将消息丢弃
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
log.info("发送消息至 exchange, 消息唯一标识: {}, 确认状态: {}, 造成原因: {}",correlationData, ack, cause);
});
rabbitTemplate.setReturnsCallback((message, replyCode, replyText, exchange, routingKey) -> {
log.error("发送消息至 queue 失败, 消息:{}, 回应码:{}, 回应信息:{}, 交换机:{}, 路由键:{}", message, replyCode, replyText, exchange, routingKey);
});
return rabbitTemplate;
}
}
生产者发送消息设置消息唯一标识
@GetMapping("sendDirectMsg")
public String sendDirectMsg() {
CorrelationData data = new CorrelationData();
data.setId("111")
Map<String, Object> map = new HashMap<>();
map.put("msg", "test_send_direct_msg");
rabbitTemplate.convertAndSend("test_direct_exchange", "test_direct_routing", map, data);
return "OK";
}
2. 消费者确认
RabbitMQ 支持消息确定 ACK 机制，消费者从 RabbitMQ 收到消息并处理完成后，返回给 RabbitMQ，RabbitMQ 收到反馈后才将此消息从队列删除
RabbitMQ 的消息确认方式有两种：自动确认和手动确认
RabbitMQ 默认是自动确认，即消息推送给消费者后，马上确认并销毁，但假如消费消息的过程中发生了异常，由于消息已经销毁，这样就会造成消息丢失
手动确认又分为肯定确认和否定确认
肯定确认：
// 第一个参数表示当前的投递标签号，相当于当前消息的 Id
// 第二个参数表示是否批量确认，true 表示批量确认当前及之前的所有消息，false表示只确认当前消息
channel.basicAck(envelope.getDeliveryTag(), false);
否定确认：
// 第一个参数表示当前的投递标签号，相当于当前消息的 Id
// 第二个参数表示是否批量拒绝，true 表示所有投递标签小于当前消息且未确认的消息都将被拒绝，false 表示仅拒绝当前消息
// 第三个参数表示被拒绝的消息是否重新放回队列，true 表示消息重新放回队列投递，false 表示丢弃该消息
channel.basicNack(envelope.getDeliveryTag(), false, true);
# rejeck 与 nack 作用相同，但不支持批量操作
channel.basicReject(envelope.getDeliveryTag(), true);
Springboot 提供了三种确认模式，配置如下：
# none：默认所有消息消费成功
# auto：根据消息处理逻辑是否抛出异常自动发送 ack（正常）或 nack（异常）
# manual：消费者手动调用 ack、nack、reject 几种方法进行确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
整合 SpringBoot 代码如下：
@Slf4j
@Component
public class MsgConfirmController {
@RabbitListener(queues = "test_confirm_queue")
public void consumerConfirm(Message message, Channel channel) throws Exception {
if(message.getBody().equals("2")) {
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
log.info("接收的消息为: {}", message);
} else {
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
log.info("未消费数据");
}
}
}
RabbitMQ 推拉模型
RabbitMQ 有两种消息处理模式：推模式和拉模式
推模式下，生产者发布消息到队列时，会立即将这条消息发送给所有订阅该队列的消费者，优点：实现实时通信，缺点：如果消费者的处理能力跟不上生产者的速度，就会在消费者处造成消息堆积，因此需要根据消费能力做流控（比如 RabbitMQ 用 QOS 来限制），RabbitMQ 默认使用推消息
拉模式下，生产者发布消息到队列时，不会立即发送消息给消费者，而是等待消费者请求消息后才发送，优点：消费端可以按照自己的处理速度来消费，缺点：消息传递存在延迟，当处理速度小于发布速度时，容易造成消息堆积在队列
SpringBoot 实现拉消息代码如下：
@Slf4j
@Component
public class RabbitMQPullConsumer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void process() {
rabbitTemplate.execute(new ChannelCallback<Object>() {
Object result;
GetResponse response;
try {
response = channel.basicGet("test_pull_queue", false);
result = new String(response.getBody(), "UTF-8");
log.info("消费者接收到消息: {}", result);
channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
} catch(Exception e) {
log.info("消费者接收消息失败", e);
if(response != null) {
try {
channel.basicAck(response.getEnvelope().getDeliveryTag(), false, true);
} catch(Exception e) {
log.info("消费者拒绝消息失败", e);
}
}
}
}
}
}
RabbitMQ 的 Channel 提供了 basicGet 方法用于拉取消息，第二个参数为是否自动 ack。这里我们需要手动调用 process 方法来拉取消息，一般来说会让一个线程负责循环拉取消息，存入一个长度有限的阻塞队列，另一个线程从阻塞队列取出消息，处理完一条则手动 Ack 一条。如果想批量拉取消息，可以连续调用 basicGet 方法拉取多条消息，处理完成之后一次性 ACK
![[AIGC]GPT模型概述](https://img2023.cnblogs.com/blog/1173617/202303/1173617-20230330015003037-631510591.png)
