0
0

RabbitMQ 学习笔记

为什么使用消息队列？

以用户下单购买商品的行为举例，在使用微服务架构时，我们需要调用多个服务，传统的调用方式是同步调用，这会存在一定的性能问题

使用消息队列可以实现异步的通信方式，相比于同步的通信方式，异步的方式可以让上游快速成功，极大提高系统的吞吐量

消息队列的使用场景有如下：

  • 异步处理：以上述用户下单购买商品为例，将多个不关联的任务放进消息队列，提高系统性能
  • 应用解耦：以上述用户下单购买商品为例，订单系统通知库存系统减库存，传统的做法是订单系统调用库存系统的接口，订单系统和库存系统高耦合，当库存系统出现故障时，订单就会失败。使用消息队列，用户下单后，订单系统完成持久化，将消息写入消息队列，返回用户下单成功。库存系统订阅下单消息，获取下单消息，进行减库存操作。就算库存系统出现故障，消息队列也能保证消息的可靠投递，不会导致系统异常
  • 流量削峰：举行秒杀活动时，为防止流量过大导致应用挂掉，服务器收到用户请求后，先写入消息队列，如果超过了消息队列长度的最大值，则直接抛弃或跳转到错误页面。秒杀业务根据消息队列中的请求信息，再做后续处理，缓解短时间的流量高峰

RabbitMQ 安装

以 ubuntu 22.04.3 为例，参考 RabbitMQ 官网提供的安装脚本

#!/bin/sh## 要想安装最新版本的 rabbitmq，可以选择 Cloudsmith 存储库下载，为此我们必须安装 apt-transport https 包sudoapt-getinstallcurl gnupg apt-transport-https -y## 获取 Cloudsmith 存储库提供的签名密钥并添加到系统中，这样这样才能使用 Cloudsmith 仓库下载包curl-1sLf"https://keys.openpgp.org/vks/v1/by-fingerprint/0A9AF2115F4687BD29803A206B73A36E6026DFCA"|sudo gpg --dearmor|sudotee /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|sudotee /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|sudotee /usr/share/keyrings/rabbitmq.9F4587F226208342.gpg > /dev/null

## 将描述 RabbitMQ 和 Erlang 包存储库的文件放在 /etc/apt/sources.list.d/ 目录下sudotee /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## 更新存储库索引sudoapt-get update -y## 安装 erlangsudoapt-getinstall-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

## 安装 rabbitmqsudoapt-getinstall 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:5672username: root
        password:123

生产者端配置队列、交换机

@ConfigurationpublicclassRabbitMqConfig{@BeanpublicQueuerabbitmqTestDirectQueue(){// Direct 队列// name：队列名称// durable：是否持久化// exclusive：是否独享，如果设置 true，则只有创建者可以使用此队列// autoDelete: 是否自动删除，也就是临时队列，当最后一个消费者断开连接就会自动删除returnnewQueue("test_direct_queue",true,false,false);}@BeanpublicDirectExchangerabbitmqTestDirectExchange(){// Direct 交换机returnnewDirectExchange("test_direct_exchange",true,false);}@BeanpublicBindingbindDirect(){returnBindingBuilder// // 绑定 Direct 队列.bind(rabbitmqTestDirectQueue())//到 Direct 交换机.to(rabbitmqTestDirectExchange())// 并设置匹配键.with("test_direct_routing");}@BeanpublicQueuerabbitmqTestFanoutQueueA(){// fanout 队列 areturnnewQueue("test_fanout_queue_a",true,false,false);}@BeanpublicQueuerabbitmqTestFanoutQueueB(){// fanout 队列 breturnnewQueue("test_fanout_queue_b",true,false,false);}@BeanpublicFanoutExchangerabbitmqTestFanoutExchange(){// Fanout 交换机returnnewFanoutExchange("test_fanout_exchange",true,false);}@BeanpublicBindingbindFanoutA(){returnBindingBuilder// 绑定 Fanout 队列 a.bind(rabbitmqTestFanoutQueueA())//到 Fanout 交换机.to(rabbitmqTestFanoutExchange());}@BeanpublicBindingbindFanoutB(){returnBindingBuilder// 绑定 Fanout 队列 b.bind(rabbitmqTestFanoutQueueB())//到 Fanout 交换机.to(rabbitmqTestFanoutExchange());}@BeanpublicQueuerabbitmqTestTopicQueueA(){// topic 队列 areturnnewQueue("test_topic_queue_a",true,false,false);}@BeanpublicQueuerabbitmqTestTopicQueueB(){// topic 队列 breturnnewQueue("test_topic_queue_b",true,false,false);}@BeanpublicTopicExchangerabbitmqTestTopicExchange(){// Topic 交换机returnnewTopicExchange("test_topic_exchange",true,false);}@BeanpublicBindingbindTopicA(){returnBindingBuilder// 绑定 Topic 队列 a.bind(rabbitmqTestTopicQueueA())//到 Topic 交换机.to(rabbitmqTestTopicExchange())// 并设置匹配键.with("a.*");}@BeanpublicBindingbindTopicB(){returnBindingBuilder// 绑定 Topic 队列 b.bind(rabbitmqTestTopicQueueB())//到 Topic 交换机.to(rabbitmqTestTopicExchange())// 并设置匹配键.with("b.*");}@BeanpublicQueuerabbitmqTestHeadersQueueA(){// headers 队列 areturnnewQueue("test_headers_queue_a",true,false,false);}@BeanpublicQueuerabbitmqTestHeadersQueueB(){// headers 队列 breturnnewQueue("test_headers_queue_b",true,false,false);}@BeanpublicHeadersExchangerabbitmqTestHeadersExchange(){// Headers 交换机returnnewHeadersExchange("test_headers_exchange",true,false);}@BeanpublicBindingbindHeadersA(){Map<String,Object> map =newHashMap<>();
      map.put("key_a1","a1");
      map.put("key_a2","a2");returnBindingBuilder// 绑定 Headers 队列 a.bind(rabbitmqTestHeadersQueueA())//到 Headers 交换机.to(rabbitmqTestHeadersExchange())// 全部匹配.whereAll(map).match();}@BeanpublicBindingbindHeadersB(){Map<String,Object> map =newHashMap<>();
      map.put("key_b1","b1");
      map.put("key_b2","b2");returnBindingBuilder// 绑定 Headers 队列 b.bind(rabbitmqTestHeadersQueueB())//到 Headers 交换机.to(rabbitmqTestHeadersExchange())// 部分匹配.whereAny(map).match();}}

生产者端发送消息

@ControllerpublicclassRabbitMQController{@AutowiredprivateRabbitTemplate rabbitTemplate;@GetMapping("sendDirectMsg")publicStringsendDirectMsg(){Map<String,Object> map =newHashMap<>();
        map.put("msg","test_send_direct_msg");
        rabbitTemplate.convertAndSend("test_direct_exchange","test_direct_routing", map);return"OK";}@GetMapping("sendFanoutMsg")publicStringsendFanoutMsg(){Map<String,Object> map =newHashMap<>();
        map.put("msg","test_send_fanout_msg");
        rabbitTemplate.convertAndSend("test_fanout_exchange","", map);return"OK";}@GetMapping("sendTopicMsgA")publicStringsendTopicMsgA(){Map<String,Object> map =newHashMap<>();
        map.put("msg","test_send_topic_msg_a");
        rabbitTemplate.convertAndSend("test_topic_exchange","a.c", map);return"OK";}@GetMapping("sendTopicMsgB")publicStringsendTopicMsgB(){Map<String,Object> map =newHashMap<>();
        map.put("msg","test_send_topic_msg_b");
        rabbitTemplate.convertAndSend("test_topic_exchange","b.c", map);return"OK";}@GetMapping("sendHeadersMsgA")publicStringsendHeadersMsgA(){MessageProperties msgp =newMessageProperties();
        msgp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        msgp.setContentType("UTF-8");Map<String,Object> map =newHashMap<>();
        map.put("key_a1","a1");
        map.put("key_a2","a2");
        msgp.getHeaders().putAll(map);Message msg =newMessage("test_send_headers_msg_a".getBytes(), msgp);
        rabbitTemplate.convertAndSend("test_headers_exchange",null, msg);return"OK";}@GetMapping("sendHeadersMsgB")publicStringsendHeadersMsgB(){MessageProperties msgp =newMessageProperties();
        msgp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        msgp.setContentType("UTF-8");Map<String,Object> map =newHashMap<>();
        map.put("key_b1","a1");
        map.put("key_b3","b3");
        msgp.getHeaders().putAll(map);Message msg =newMessage("test_send_headers_msg_b".getBytes(), msgp);
        rabbitTemplate.convertAndSend("test_headers_exchange",null, msg);return"OK";}}

消费端引入的依赖和配置与生产端一样，接收消息

/**
 * Direct 队列消费者
 */@Slf4j@Component@RabbitListener(queues ={"test_direct_queue"})publicclassRabbitMqDirectReceiver{@RabbitHandlerpublicvoidprocess(Map<String,Object> map){
    log.info("消费者接收到消息: {}", map.toString());}}

/**
 * Fanout 队列消费者 a
 */@Slf4j@Component@RabbitListener(queues ={"test_fanout_queue_a"})publicclassRabbitMqFanoutAReceiver{@RabbitHandlerpublicvoidprocess(Map<String,Object> map){
    log.info("消费者接收到消息: {}", map.toString());}}

/**
 * Fanout 队列消费者 b
 */@Slf4j@Component@RabbitListener(queues ={"test_fanout_queue_b"})publicclassRabbitMqFanoutBReceiver{@RabbitHandlerpublicvoidprocess(Map<String,Object> map){
    log.info("消费者接收到消息: {}", map.toString());}}

/**
 * Topic 队列消费者 a
 */@Slf4j@Component@RabbitListener(queues ={"test_topic_queue_a"})publicclassRabbitMqTopicAReceiver{@RabbitHandlerpublicvoidprocess(Map<String,Object> map){
    log.info("消费者接收到消息: {}", map.toString());}}

/**
 * Topic 队列消费者 b
 */@Slf4j@Component@RabbitListener(queues ={"test_topic_queue_b"})publicclassRabbitMqTopicBReceiver{@RabbitHandlerpublicvoidprocess(Map<String,Object> map){
    log.info("消费者接收到消息: {}", map.toString());}}

/**
 * Topic 队列消费者 a
 */@Slf4j@Component@RabbitListener(queues ={"test_topic_queue_a"})publicclassRabbitMqTopicAReceiver{@RabbitHandlerpublicvoidprocess(Map<String,Object> map){
    log.info("消费者接收到消息: {}", map.toString());}}

/**
 * Headers 队列消费者 a
 */@Slf4j@ComponentpublicclassRabbitMqHeadersAReceiver{@RabbitListener(queuesToDeclare =@Queue("test_headers_queue_a"))publicvoidprocess(Message msg)throwsException{MessageProperties msgp = message.getMessageProperties();String contentType = msgp.getContentType();
    log.info("消费者接收到消息: {}",newString(message.getBody(), contentType));}}

/**
 * Headers 队列消费者 b
 */@Slf4j@ComponentpublicclassRabbitMqHeadersBReceiver{@RabbitListener(queuesToDeclare =@Queue("test_headers_queue_b"))publicvoidprocess(Message msg)throwsException{MessageProperties msgp = message.getMessageProperties();String contentType = msgp.getContentType();
    log.info("消费者接收到消息: {}",newString(message.getBody(), contentType));}}

RabbitMQ 消息确认机制

消费者从队列中获取到消息之后，在处理消息时出现异常，那这条正在处理的消息就没有完成消息消费，数据就会丢失。生产者同样如此，生产者发消息给交换机，也不能保证消息准确发送过去了

RabbitMQ 的消息确认分为两部分：

  • 消息发送确认：用来确认生产者是否成功将消息投递到队列
  • 消息接收确认：用来确认消费者是否成功接收到消息
1. 生产端确认

RabbitMQ 提供了两种机制，用于告知生产端是否发送消息成功：

  • publisher-confirm：消息投递交换机，返回成功/失败信息
  • publisher-return：消息投递交换机成功，但路由到队列失败，返回失败信息

生产端配置如下：

spring:rabbitmq:# 开启 publisher-confirmpublisher-confirm-type: correlated
    # 开启 publisher-returnpublisher-returns:true

publish-confirm-type

有三个值：

  • none：禁用发布确认模式，默认值
  • simple：同步等待 confirm 结果，直到超时
  • correlated：异步通知 confirm 结果，需要定义回调函数 ConfirmCallback

生产端配置 ConfirmCallback 函数和 ReturnCallback 函数

@Slf4j@ConfigurationpublicclassProviderCallBackConfig{@AutowiredprivateCachingConnectionFactory factory;@BeanpublicRabbitTemplaterabbitTemplate(){RabbitTemplate rabbitTemplate =newRabbitTemplate(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")publicStringsendDirectMsg(){CorrelationData data =newCorrelationData();
    data.setId("111")Map<String,Object> map =newHashMap<>();
    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@ComponentpublicclassMsgConfirmController{@RabbitListener(queues ="test_confirm_queue")publicvoidconsumerConfirm(Message message,Channel channel)throwsException{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@ComponentpublicclassRabbitMQPullConsumer{@AutowiredprivateRabbitTemplate rabbitTemplate;publicvoidprocess(){
    rabbitTemplate.execute(newChannelCallback<Object>(){Object result;GetResponse response;try{
        response = channel.basicGet("test_pull_queue",false);
        result =newString(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

RabbitMQ 持久化

如果RabbitMQ 服务宕机，就会导致消息丢失，因此我们需要实现持久化，无论是 RabbitMQ 服务重启、崩溃，也不会丢失消息

RabbitMQ 持久化包含三个方面：exchange 持久化、queue 持久化、message 持久化

创建 exchange 时，设置 durable 参数为 true，则该 exchange 做持久化，重启 rabbitmq 服务器，该 exchange 不会消失，durable 的默认值为 true

@ConfigurationpublicclassRabbitMqConfig{@BeanpublicDirectExchangerabbitmqTestDirectExchange(){// Direct 交换机// 设置第二个参数 durable 为 true，开启持久化returnnewDirectExchange("test_direct_exchange",true,false);}}

创建 queue 开启持久化也是同理

@ConfigurationpublicclassRabbitMqConfig{@BeanpublicQueuerabbitmqTestDirectQueue(){// Direct 队列// 设置第二个参数 durable 为 true，开启持久化returnnewQueue("test_direct_queue",true,false,false);}}

实现消息持久化，需要在生产者推送消息时进行设置，使用 RabbitTemplate 推送消息默认都是开启持久化的

publicvoidsendMsg(){MessageProperties msgp =newMessageProperties();// MessageDeliveryMode.PERSISTENT 表示设置为持久化
    msgp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);Message msg =newMessage("test__msg".getBytes(), msgp);
    rabbitTemplate.convertAndSend("test__exchange",null, msg);}

将消息标记为持久化并不能完全保证不丢消息，尽管它告诉 RabbitMQ 将消息保存到磁盘，但依然存在当消息刚准备存储在磁盘，但是还没有完全存储完，消息还在缓存的一个间隔点，此时如果 RabbitMQ 宕机，一样会造成数据丢失

RabbitMQ 消息顺序性

RabbitMQ 并不能直接保证消息的消费顺序，因为它是将消息发送到多个消费者并行处理。一般来说，我们会将需要保证顺序的消息放到同一个消息队列，然后只用一个消费者去消费该队列，同一个队列的消息一定是有序的

如果消息量很大，全部放在一个队列会增加系统压力，这时我们可以考虑消息分区：根据消息中的特定属性（例如 id）将消息分发到不同队列，每个队列对应一个消费者，每个队列的消息是有序的。具体操作可以通过路由键来实现：创建三个队列，每个队列都对应不同的路由键，每个消息可以根据 id 设置不同的路由键（例如用 id 对路由键的数量取模），消费者通过绑定指定的队列和路由键来接收消息，这样每个消费者只会接收到其关注的特定路由键的消息，从而实现了消息的分区

无论是使用单个队列还是消息分区，都会有一个问题：如果消费者挂了，那么就无法继续消费了。如果消费者服务以集群部署，可以指定其中一个消费者消费消息来保证有序性，如果这个消费者挂了，那如何保证消息继续正常消费呢？

  1. 一个队列绑定多个消费者，每个消费者消费消息时要先获取分布式锁，否则不能消费
  2. 一个队列绑定多个消费者，每个消费者指定不同的优先级，RabbitMQ 会优先将消息发送给优先级高的消费者，实现按优先级有序消费
  3. 使用 Fanout 交换机，为每个消费者创建一个队列，这样子每个消费者都会同时收到消息，消费者端保证消费接口幂等性，也就是保证消息不会重复消费
  4. 使用 Fanout 交换机，为每个消费者创建一个队列，这样子每个消费者都会同时收到消息，消费者通过访问 RabbitMQ 的接口获取这所有消费者的 ip 并保存在集合中，消费者监听到消息后，判断自己的 ip 是否是 ip 集合中的最小值，是则消费，否则抛弃。如果最小 ip 的消费者宕机，则集合中将剔除宕机 ip，后续的消息仍然可以从剩余的 ip 中寻找最小值消费
标签： rabbitmq
本文转载自: https://blog.csdn.net/CSDN_handsome/article/details/136326639
版权归原作者 低吟不作语 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ 学习笔记”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

机器人技术在安全领域的重要作用

安全学习记录——网络篇（一）

图腾柱PFC：HP1010为您的电动两轮车之旅提供绿色，高效，安全的动力

单元测试数据库回滚问题

这才是单元测试，之前我们都错了

【C语言基础】：字符函数和字符串函数

ChatGPT-Next-Web SSRF漏洞+XSS漏洞复现(CVE-2023-49785)

文章导航