2024-02-16发表2024-08-11更新 Xiamu SpringCloud / RabbitMQ5 分钟读完 (大约752个字)

Spring AMQP(3.1.1)设置ConfirmCallback和ReturnsCallback

环境如下

	Version
SpringBoot	3.2.1
spring-amqp	3.1.1
RabbitMq	3-management

一、起因

老版本的spring-amqp在CorrelationData上设置ConfirmCallback。但是今天却突然发现correlationData.getFuture()没有addCallback函数了。

查询文档和帖子后，发现ConfirmCallback和ReturnsCallback都需要在RabbitTemplate中设置，同时ConfirmCallback中默认无法得到消息内容，如果想在ConfirmCallback中把消息内容存到数据库等地方进行记录，怎么办呢？

参考手册

二、代码

1. 定义exchange和queue

language-java

@Slf4j
@Configuration
public class PayNotifyConfig{
   

    //交换机
    public static final String PAYNOTIFY_EXCHANGE_FANOUT = "paynotify_exchange_fanout";
    //支付通知队列
    public static final String PAYNOTIFY_QUEUE = "paynotify_queue";
    //支付结果通知消息类型
    public static final String MESSAGE_TYPE = "payresult_notify";


    //声明交换机，且持久化
    @Bean(PAYNOTIFY_EXCHANGE_FANOUT)
    public FanoutExchange paynotify_exchange_fanout() {
   
        // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
        return new FanoutExchange(PAYNOTIFY_EXCHANGE_FANOUT, true, false);
    }
    //支付通知队列,且持久化
    @Bean(PAYNOTIFY_QUEUE)
    public Queue paynotify_queue() {
   
        return QueueBuilder.durable(PAYNOTIFY_QUEUE).build();
    }

    //交换机和支付通知队列绑定
    @Bean
    public Binding binding_paynotify_queue(@Qualifier(PAYNOTIFY_QUEUE) Queue queue, @Qualifier(PAYNOTIFY_EXCHANGE_FANOUT) FanoutExchange exchange) {
   
        return BindingBuilder.bind(queue).to(exchange);
    }
}

2. RabbitTemplate

在上面的类中继续添加RabbitTemplate ，并设置ConfirmCallback和ReturnsCallback。

language-java

@Bean
   public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
  
       final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
       //设置confirm callback
       rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
  
           String body = "1";
           if (correlationData instanceof EnhancedCorrelationData) {
  
               body = ((EnhancedCorrelationData) correlationData).getBody();
           }
           if (ack) {
  
               //消息投递到exchange
               log.debug("消息发送到exchange成功:correlationData={},message_id={} ", correlationData, body);
               System.out.println("消息发送到exchange成功:correlationData={},message_id={}"+correlationData+body);
           } else {
  
               log.debug("消息发送到exchange失败:cause={},message_id={}",cause, body);
               System.out.println("消息发送到exchange失败:cause={},message_id={}"+cause+body);
           }
       });
       
       //设置return callback
       rabbitTemplate.setReturnsCallback(returned -> {
  
           Message message = returned.getMessage();
           int replyCode = returned.getReplyCode();
           String replyText = returned.getReplyText();
           String exchange = returned.getExchange();
           String routingKey = returned.getRoutingKey();
           // 投递失败，记录日志
           log.error("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",
                   replyCode, replyText, exchange, routingKey, message.toString());
       });
       return rabbitTemplate;
   }

3. EnhancedCorrelationData

原始的CorrelationData，目前已经无法从中获取消息内容，也就是说现在的ConfirmCallback无法获取到消息的内容，因为设计上只关注是否投递到exchange成功。如果需要在ConfirmCallback中获取消息的内容，需要扩展这个类，并在发消息的时候，放入自定义数据。

language-java

public class EnhancedCorrelationData extends CorrelationData {
   
    private final String body;

    public EnhancedCorrelationData(String id, String body) {
   
        super(id);
        this.body = body;
    }

    public String getBody() {
   
        return body;
    }
}

4. 发送消息

在EnhancedCorrelationData把消息本身放进去，或者如果你有表记录消息，你可以只放入其id。这样触发ConfirmCallback的时候，就可以获取消息内容。

language-java

public void notifyPayResult() {
 
String message = "TEST Message";
      Message message1 = MessageBuilder.withBody(message.getBytes(StandardCharsets.UTF_8))
              .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
              .build();
      CorrelationData correlationData = new EnhancedCorrelationData(UUID.randomUUID().toString(), message.toString());
      rabbitTemplate.convertAndSend(PayNotifyConfig.PAYNOTIFY_EXCHANGE_FANOUT,"", message1, correlationData);
  }

2024-01-17发表2024-08-11更新 Xiamu SpringCloud / RabbitMQ10 分钟读完 (大约1460个字)

RabbitMQ常见问题之高可用

一、集群分类

RabbitMQ的是基于Erlang语言编写,而Erlang又是一个面向并发的语言,天然支持集群模式。RabbitMQ的集群有两
种模式：

普通集群：是一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群的并发能力。
镜像集群：是一种主从集群，普通集群的基础上，添加了主从备份功能，提高集群的数据可用性。

镜像集群虽然支持主从,但主从同步并不是强一致的,某些情况下可能有数据丢失的风险。因此在RabbitMQ的3.8版本
以后,推出了新的功能——仲裁队列来代替镜像集群,底层采用Raft协议确保主从的数据一致性。

二、普通集群搭建

1. 准备

建立如下文件夹结构

language-txt

./cluster/
├── docker-compose.yml
├── mq1
│   ├── .erlang.cookie
│   └── conf
│       └── rabbitmq.conf
├── mq2
│   ├── .erlang.cookie
│   └── conf
│       └── rabbitmq.conf
└── mq3
    ├── .erlang.cookie
    └── conf
        └── rabbitmq.conf

2. 配置

rabbitmq.conf都写入以下内容

language-shell

loopback_users.guest = false
listeners.tcp.default = 5672
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 = rabbit@mq1
cluster_formation.classic_config.nodes.2 = rabbit@mq2
cluster_formation.classic_config.nodes.3 = rabbit@mq3
vm_memory_high_watermark.absolute = 524288000

.erlang.cookie都写入以下内容

language-txt

1	SUGWXEQPRCPYJAVYPNZY

集群的所有节点的.erlang.cookie需要保持一致才能互相信任，具体内容并不固定，可以随便新建一个rabbitmq容器去查看其.erlang.cookie然后复制使用即可。

docker-compose.yml写入以下内容

language-yml

version: "3.8"

networks:
  rabbitmq-normal-cluster:
    driver: bridge
    ipam:
      driver: default
      config:
        - subnet: 172.30.3.0/24

services:
  mq1:
    container_name: mq1
    hostname: mq1
    image: rabbitmq:3-management
    environment:
      - RABBITMQ_DEFAULT_USER=rabbitmq
      - RABBITMQ_DEFAULT_PASS=rabbitmq
    volumes:
      - ./mq1/conf/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf
      - ./mq1/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie:ro
    ports:
      - "8071:5672"
      - "8081:15672"
    networks:
      rabbitmq-normal-cluster:
        ipv4_address: 172.30.3.11

  mq2:
    container_name: mq2
    hostname: mq2
    image: rabbitmq:3-management
    environment:
      - RABBITMQ_DEFAULT_USER=rabbitmq
      - RABBITMQ_DEFAULT_PASS=rabbitmq
    volumes:
      - ./mq2/conf/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf
      - ./mq2/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie:ro
    ports:
      - "8072:5672"
      - "8082:15672"
    networks:
      rabbitmq-normal-cluster:
        ipv4_address: 172.30.3.12

  mq3:
    container_name: mq3
    hostname: mq3
    image: rabbitmq:3-management
    environment:
      - RABBITMQ_DEFAULT_USER=rabbitmq
      - RABBITMQ_DEFAULT_PASS=rabbitmq
    volumes:
      - ./mq3/conf/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf
      - ./mq3/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie:ro
    ports:
      - "8073:5672"
      - "8083:15672"
    networks:
      rabbitmq-normal-cluster:
        ipv4_address: 172.30.3.13

3. 运行

language-bash

1	docker-compose -p rabbitmq-c up -d

三、镜像集群

1. 介绍

镜像模式的配置有3种模式：

ha-mode	ha-params	效果
准确模式`exactly`	队列的副本量`count`	集群中队列副本（主服务器和镜像服务器之和）的数量。`count`如果为1意味着单个副本：即队列主节点。`count`值为2表示2个副本：1个队列主和1个队列镜像。换句话说：`count` = 镜像数量 + 1。如果群集中的节点数少于`count`，则该队列将镜像到所有节点。如果有集群总数大于`count`+1，并且包含镜像的节点出现故障，则将在另一个节点上创建一个新的镜像。
`all`	(`none`)	队列在群集中的所有节点之间进行镜像。队列将镜像到任何新加入的节点。镜像到所有节点将对所有群集节点施加额外的压力，包括网络I / O，磁盘I / O和磁盘空间使用情况。推荐使用`exactly`，设置副本数为（`N` / 2 +1）。
`nodes`	`node names`	指定队列创建到哪些节点，如果指定的节点全部不存在，则会出现异常。如果指定的节点在集群中存在，但是暂时不可用，会创建节点到当前客户端连接到的节点。

2. 启用方式

三种模式启动方式分别如下，基于普通集群之上，命令均需要在单个容器内部执行。

language-bash

1	rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

language-bash

1	rabbitmqctl set_policy ha-all "^all\." '{"ha-mode":"all"}'

language-bash

1	rabbitmqctl set_policy ha-nodes "^nodes\." '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}'

3. 测试

这里以exactly为例，在mq1中执行rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'后，所有前缀为two的queue都会有1个主queue和1个副本。

language-txt

1
2
3

root@mq1:/# rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'
Setting policy "ha-two" for pattern "^two\." to "{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}" with priority "0" for vhost "/" ...
root@mq1:/#

来到localhost:8081管理页，找到admin->policies可以看到策略生效。

来新建一个two.test.queue，可以看到这是一个拥有副本的queue。

四、仲裁队列

1. 介绍

仲裁队列：仲裁队列是3.8版本以后才有的新功能，用来替代镜像队列，具备下列特征：

与镜像队列一样，都是主从模式，支持主从数据同步
使用非常简单，没有复杂的配置
主从同步基于Raft协议，强一致

上一章中想要一个镜像队列还要执行各种命令，遵循规定，现在不用了。

2. 创建

java使用目前只能基于@Bean创建

language-java

@Bean
public Queue quorumQueue(){
   
    return QueueBuilder
            .durable("quorum.queue2")
            .quorum()
            .build();
}

五、Java连接RabbitMQ集群方式

Java使用RabbitMQ集群application.yml中需要修改address

language-yml

spring:
  rabbitmq:
    host: localhost # rabbitMQ的ip地址
    port: 5672 # 端口

language-yml

1
2
3

spring:
  rabbitmq:
	addresses: localhost:8071, localhost:8072, localhost:8073

2024-01-17发表2024-08-11更新 Xiamu SpringCloud / RabbitMQ2 分钟读完 (大约314个字)

RabbitMQ常见问题之消息堆积

一、介绍

当生产者发送消息的速度超过了消费者处理消息的速度,就会导致队列中的消息堆积,直到队列存储消息达到上限。最
早接收到的消息，可能就会成为死信，会被丢弃，这就是消息堆积问题。

解决消息堆积有三种种思路：

增加更多消费者，提高消费速度
在消费者内开启线程池加快消息处理速度
扩大队列容积，提高堆积上限

从RabbitMQ的3.6.0版本开始，就增加了Lazy Queues的概念，也就是惰性队列。惰性队列的特征如下：

接收到消息后直接存入磁盘而非內存
消费者要消费消息时才会从磁盘中读取并加载到内存
支持数百万条的消息存储

二、使用惰性队列

1. 基于@Bean

language-java

@Bean
    public Queue lazyQueue(){
   
        return QueueBuilder
                .durable("lazy.queue")
                .lazy()
                .build();
    }

2. 基于@RabbitListener

language-java

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "lazy.queue", arguments = @Argument(name = "x-queue-mode", value = "lazy")),
            exchange = @Exchange(name = "simple.exchange"),
            key = "lazy"
    ))
    public void listenLazyExchange(String msg){
   
//        log.info("消费者接收到lazy.queue的消息：【" + msg + "】");
    }

2024-01-17发表2024-08-11更新 Xiamu SpringCloud / RabbitMQ6 分钟读完 (大约885个字)

RabbitMQ常见问题之延迟消息

一、死信交换机

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
消息是一个过期消息，超时无人消费
要投递的队列消息堆积满了，最早的消息可能成为死信

如果该队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而
这个交换机称为死信交换机(Dead Letter Exchange,简称DLX)。

二、TTL

如果message和queue都有ttl，采用更小的一方。

1. Queue指定死信交换机并设置TTL

language-java

@Configuration
public class CommonConfig {
   
    @Bean
    public DirectExchange ttlExchange(){
   
        return new DirectExchange("ttl.direct");
    }

    @Bean
    public Queue ttlQueue(){
   
        return QueueBuilder
                .durable("ttl.queue")
                .ttl(10000)
                .deadLetterExchange("dl.direct")
                .deadLetterRoutingKey("dl")
                .build();
    }

    @Bean
    public Binding ttlBinding(){
   
        return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
    }
}

2. 消息设置TTL

language-java

@Test
public void testTTLMessage(){
   
    Message message = MessageBuilder.withBody("hello ttl".getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .setExpiration("5000")
            .build();
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message);
    log.info("ttl消息已发送");
}

借助TTL机制可以用死信交换机模拟延迟队列，但是设计上比较牵强，性能不好。

三、延迟队列

这是官方提供的一些额外插件
https://www.rabbitmq.com/community-plugins.html

下载其中的DelayExchange插件，把.ez文件挂载到RabbitMQ容器的/plugins目录下，然后进入容器，执行

language-bash

1	rabbitmq-plugins enable rabbitmq_delayed_message_exchange

language-txt

root@7c4ba266e5bc:/# rabbitmq-plugins enable rabbitmq_delayed_message_exchange
Enabling plugins on node rabbit@7c4ba266e5bc:
rabbitmq_delayed_message_exchange
The following plugins have been configured:
  rabbitmq_delayed_message_exchange
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_prometheus
  rabbitmq_web_dispatch
Applying plugin configuration to rabbit@7c4ba266e5bc...
The following plugins have been enabled:
  rabbitmq_delayed_message_exchange

started 1 plugins.

1. SpringAMQP创建延迟队列

基于@RabbitListener或者基于@Bean都可以。

language-java

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "delay.queue"),
        exchange = @Exchange(name = "delay.direct", delayed = "true"),
        key = "delay"
))
public void listenDelayExchange(String msg){
   
    log.info("消费者接收到delay.queue的延迟消息：【" + msg + "】");
}

2. 设置消息延迟

这个插件只能在消息上设置延迟时间，没有队列设置延迟时间的概念，不过都是一样的。
message要在Header上添加一个x-delay。

language-java

@Test
public void testDelayMessage(){
   
    Message message = MessageBuilder.withBody("hello delay".getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .setHeader("x-delay", 5000)
            .build();
    // confirm callback
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);
    log.info("发送消息成功");
}

3. 测试

直接运行测试，可能会报错，因为rabbitmq意识到消息到了exchange却没有立即到queue，被认为错误，回调returnback，所以我们在ReturnCallBack中绕过这个限制。

language-java

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {
   

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
   
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey)->{
   
            //check if is delay message
            if (message.getMessageProperties().getReceivedDelay() != null && message.getMessageProperties().getReceivedDelay() > 0) {
   
                return;
            }
            log.error("消息发送到queue失败，replyCode={}, reason={}, exchange={}, routeKey={}, message={}",
                    replyCode, replyText, exchange, routingKey, message.toString());
        });
    }
}

运行Test测试，可以看到Test方面，消息发送的时间为21:09:13

language-txt

1
2

21:09:13:516  INFO 25468 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#2063c53e:0/SimpleConnection@6415f61e [delegate=amqp://rabbitmq@127.0.0.1:5672/, localPort= 62470]
21:09:13:557  INFO 25468 --- [           main] cn.itcast.mq.spring.SpringAmqpTest       : 发送消息成功

listener方面消息消费的时间为21:09:18，刚好5s。

language-txt

1
2

21:08:31:952  INFO 19532 --- [           main] cn.itcast.mq.ConsumerApplication         : Started ConsumerApplication in 1.735 seconds (JVM running for 2.357)
21:09:18:583  INFO 19532 --- [ntContainer#0-1] c.i.mq.listener.SpringRabbitListener     : 消费者接收到delay.queue的延迟消息：【hello delay】

2024-01-17发表2024-08-11更新 Xiamu SpringCloud / RabbitMQ11 分钟读完 (大约1690个字)

RabbitMQ常见问题之消息可靠性

一、介绍

MQ的消息可靠性，将从以下四个方面展开并实践：

生产者消息确认
消息持久化
消费者消息确认
消费失败重试机制

二、生产者消息确认

对于publisher，如果message到达exchange与否，rabbitmq提供publiser-comfirm机制，如果message达到exchange但是是否到达queue，rabbitmq提供publisher-return机制。这两种机制在代码中都可以通过配置来自定义实现。

以下操作都在publisher服务方完成。

1. 引入依赖

language-xml

<dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-amqp</artifactId>
      </dependency>

language-yaml

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true

配置说明：
publish-confirm-type:开启publisher-confirm，这里支持两种类型：

simple：同步等待confirm结果，直到超时
correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback

publish-returns：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallback
template.mandatory：定义消息路由失败时的策略。true，则调用ReturnCallback; false，则直接丢弃消息

2. 配置ReturnCallBack

每个RabbitTemplate只能配置一个ReturnCallBack，所以直接给IoC里面的RabbitTemplate配上，所有人都统一用。
新建配置类，实现ApplicationContextAware 接口，在接口中setReturnCallback。

language-java

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {
   

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
   
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey)->{
   
            //check if is delay message
            if (message.getMessageProperties().getReceivedDelay() != null && message.getMessageProperties().getReceivedDelay() > 0) {
   
                return;
            }
            log.error("消息发送到queue失败，replyCode={}, reason={}, exchange={}, routeKey={}, message={}",
                    replyCode, replyText, exchange, routingKey, message.toString());
        });
    }
}

3. 配置ConfirmCallBack

ConfirmCallBack在message发送时配置，每个message都可以有自己的ConfirmCallBack。

language-java

@Test
    public void testSendMessage2SimpleQueue() throws InterruptedException {
   
        String message = "hello, spring amqp!";
        // confirm callback
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        correlationData.getFuture().addCallback(
                result -> {
   
                    if (result.isAck()){
   
                        log.debug("消息到exchange成功, id={}", correlationData.getId());
                    }else {
   
                        log.error("消息到exchange失败, id={}", correlationData.getId());
                    }
                },
                throwable -> {
   
                    log.error("消息发送失败", throwable);
                }
        );

        rabbitTemplate.convertAndSend("amq.topic", "simple.test", message, correlationData);
    }

4. 测试

将消息发送到一个不存在的exchange，模拟消息达到exchange失败，触发ConfirmCallBack，日志如下。

language-txt

1
2

18:22:03:913 ERROR 23232 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'aamq.topic' in vhost '/', class-id=60, method-id=40)
18:22:03:915 ERROR 23232 --- [nectionFactory1] cn.itcast.mq.spring.SpringAmqpTest       : 消息到exchange失败, id=0c0910a3-7937-43ea-9606-e5bbcdda0b5c

将消息发送到一个存在的exchange，但routekey异常，模拟消息到达exchange但没有到达queue，触发ConfirmCallBack和ReturnCallBack，日志如下。

language-txt

1
2
3

18:27:22:757  INFO 20184 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#7428de63:0/SimpleConnection@6d60899e [delegate=amqp://rabbitmq@127.0.0.1:5672/, localPort= 53662]
18:27:22:797 DEBUG 20184 --- [ 127.0.0.1:5672] cn.itcast.mq.spring.SpringAmqpTest       : 消息到exchange成功, id=5fbdaaa1-5f20-4683-bdfa-bd71cd6afd11
18:27:22:796 ERROR 20184 --- [nectionFactory1] cn.itcast.mq.config.CommonConfig         : 消息发送到queue失败，replyCode=312, reason=NO_ROUTE, exchange=amq.topic, routeKey=simplee.test, message=(Body:'hello, spring amqp!' MessageProperties [headers={spring_returned_message_correlation=5fbdaaa1-5f20-4683-bdfa-bd71cd6afd11}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])

三、消息持久化

新版本的SpringAMQP默认开启持久化。RabbitMQ本身并不默认开启持久化。

队列持久化，通过QueueBuilder构建持久化队列，比如

language-java

@Bean
   public Queue simpleQueue(){
  
       return QueueBuilder
               .durable("simple.queue")
               .build();
   }

消息持久化，在发送时可以设置，比如

language-java

@Test
public void testDurableMessage(){
   
    Message message = MessageBuilder.withBody("hello springcloud".getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .build();
    rabbitTemplate.convertAndSend("simple.queue", message);
}

四、消费者消息确认

消费者消息确认是指，consumer收到消息后会给rabbitmq发送回执来确认消息接收状况。

SpringAMQP允许配置三种确认模式:

manual：手动ack，需要在业务代码结束后，调用api发送ack。
auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack
none:关闭ack, MQ假定消费者获取消息后会成功处理,因此消息投递后立即被删除

language-yaml

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1
        acknowledge-mode: auto # manual auto none

但是auto有个很大的缺陷，因为rabbitmq会自动不断给有问题的listen反复投递消息，导致不断报错，所以建议使用下一章的操作。

五、消费失败重试机制

当消费者出现异常后,消息会不断requeue (重新入队)到队列,再重新发送给消费者,然后再次异常,再次requeue
，无限循环，导致mq的消息处理飙升，带来不必要的压力。

我们可以利用Spring的retry机制,在消费者出现异常时利用本地重试,而不是无限制的requeue到mq队列。

1. 引入依赖

language-yml

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000 #初识的失败等待时长为1秒
          multiplier: 2 # 下次失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数
          stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

2. 配置重试次数耗尽策略

我们采用RepublishMessageRecoverer。
定义用于接收失败消息的exchange，queue以及它们之间的bindings。

然后定义MessageRecoverer，比如

language-java

@Component
public class ErrorMessageConfig {
   
    @Bean
    public MessageRecoverer republishMessageRecover(RabbitTemplate rabbitTemplate){
   
        return new RepublishMessageRecoverer(rabbitTemplate, "error.exchange", "error");
    }
}

3. 测试

定义处理异常消息的exchange和queue，比如

language-java

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "error.queue"),
        exchange = @Exchange(name = "error.exchange"),
        key = "error"
))
public void listenErrorQueue(String msg){
   
    log.info("消费者接收到error.queue的消息：【" + msg + "】");
}

定义如下一个listener，来模拟consumer处理消息失败触发消息重试。

language-java

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "simple.queue"),
        exchange = @Exchange(name = "simple.exchange"),
        key = "simple"
))
public void listenSimpleQueue(String msg) {
   
    log.info("消费者接收到simple.queue的消息：【" + msg + "】");
    System.out.println(1/0);
    log.info("consumer handle message success");
}

写一个简单的测试，往simple.exchange发送消息，比如

language-java

@Test
public void testSendMessageSimpleQueue() throws InterruptedException {
   
    String message = "hello, spring amqp!";
    rabbitTemplate.convertAndSend("simple.exchange", "simple", message);
}

运行测试，consumer得到以下日志

language-txt

18:51:10:164  INFO 24072 --- [ntContainer#0-1] c.i.mq.listener.SpringRabbitListener     : 消费者接收到simple.queue的消息：【hello, spring amqp!】
18:51:11:167  INFO 24072 --- [ntContainer#0-1] c.i.mq.listener.SpringRabbitListener     : 消费者接收到simple.queue的消息：【hello, spring amqp!】
18:51:13:168  INFO 24072 --- [ntContainer#0-1] c.i.mq.listener.SpringRabbitListener     : 消费者接收到simple.queue的消息：【hello, spring amqp!】
18:51:13:176  WARN 24072 --- [ntContainer#0-1] o.s.a.r.retry.RepublishMessageRecoverer  : Republishing failed message to exchange 'error.exchange' with routing key error
18:51:13:181  INFO 24072 --- [ntContainer#1-1] c.i.mq.listener.SpringRabbitListener     : 消费者接收到error.queue的消息：【hello, spring amqp!】

可以看到spring尝试2次重发，一共3次，第一次间隔1秒，第二次间隔2秒，重试次数耗尽，消息被consumer传入error.exchange，注意，是consumer传的，不是simple.queue。

2024-01-14发表2024-08-11更新 Xiamu SpringCloud / RabbitMQ3 分钟读完 (大约474个字)

Docker运行RabbitMQ并使用SpringAMQP操作

一、RabbitMQ运行

拉取docker镜像

language-bash

1	docker pull rabbitmq:3-management

基础运行命令

language-bash

docker run \
	-e RABBITMQ_DEFAULT_USER=rabbitmq \
	-e RABBITMQ_DEFAULT_PASS=rabbitmq \
	--name rabbitmq \
	-p 15672:15672 \
	-p 5672:5672 \
	-d \
	rabbitmq:3-management

15672是网页后台管理系统，5672是给服务用的。

官方入门教程可以看这里RabbitMQ Tutorials — RabbitMQ

二、整合SpringAMQP

1. 引入依赖

language-xml

<dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-amqp</artifactId>
      </dependency>

language-yml

spring:
  rabbitmq:
    host: localhost # rabbitMQ的ip地址
    port: 5672 # rabbitMQ服务端口
    username: rabbitmq
    password: rabbitmq

三、测试

这边采用常用的消费者-生产者模型，使用默认的Direct类型exchange。不懂的可以先继续学习rabbitmq再来实践。

1. 消费者

在消费者服务随便新建一个listener

language-java

@Slf4j
@Component
public class SpringRabbitListener {
   

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "simple.queue"),
            exchange = @Exchange(name = "simple.exchange"),
            key = "simple"
    ))
    public void listenSimpleQueue(String msg) {
   
        log.info("消费者接收到simple.queue的消息：【" + msg + "】");
    }
}

2. 生产者

在生产者服务的Test模块新建一个测试

language-java

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
   
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue() throws InterruptedException {
   
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend("simple.exchange", "simple", message);
    }
}

3. 运行

先启动消费者。
登录http://localhost:15672/，可以看到simple.exchange和simple.queue已被创建。

然后启动测试testSendMessage2SimpleQueue，出现类似以下日志，消息发送成功。

language-txt

1
2

12:17:43:771  INFO 21064 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [localhost:5672]
12:17:43:808  INFO 21064 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#50f40653:0/SimpleConnection@536d97f8 [delegate=amqp://rabbitmq@127.0.0.1:5672/, localPort= 59641]

消费者出现类似以下日志，收到消息。

language-txt

1
2
3

12:17:31:074  INFO 8924 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#2a27cb34:0/SimpleConnection@671facee [delegate=amqp://rabbitmq@127.0.0.1:5672/, localPort= 59634]
12:17:31:141  INFO 8924 --- [           main] cn.itcast.mq.ConsumerApplication         : Started ConsumerApplication in 1.011 seconds (JVM running for 1.462)
12:17:43:848  INFO 8924 --- [ntContainer#0-1] c.i.mq.listener.SpringRabbitListener     : 消费者接收到simple.queue的消息：【hello, spring amqp!】

一、起因

二、代码

1. 定义exchange和queue

2. RabbitTemplate

3. EnhancedCorrelationData

4. 发送消息

一、集群分类

二、普通集群搭建

1. 准备

2. 配置

3. 运行

三、镜像集群

1. 介绍

2. 启用方式

3. 测试

四、仲裁队列

1. 介绍

2. 创建

五、Java连接RabbitMQ集群方式

一、介绍

二、使用惰性队列

1. 基于@Bean

2. 基于@RabbitListener

一、死信交换机

二、TTL

1. Queue指定死信交换机并设置TTL

2. 消息设置TTL

三、延迟队列

1. SpringAMQP创建延迟队列

2. 设置消息延迟

3. 测试

一、介绍

二、生产者消息确认

1. 引入依赖

2. 配置ReturnCallBack

3. 配置ConfirmCallBack

4. 测试

三、消息持久化

四、消费者消息确认

五、消费失败重试机制

1. 引入依赖

2. 配置重试次数耗尽策略

3. 测试

一、RabbitMQ运行

二、整合SpringAMQP

1. 引入依赖

三、测试

1. 消费者

2. 生产者

3. 运行

分类

标签

最新文章

归档