springboot-rabbitmq(二)_rabbitmq channel.basicack 报错

pom

<!--amqp高级消息队列的依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
1
2
3
4
5

配置文件

#rabbitmq服务器ip
spring.rabbitmq.host=127.0.0.1
#rabbitmq的端口
spring.rabbitmq.port=5672
#用户名
spring.rabbitmq.username=guest
#密码
spring.rabbitmq.password=guest
#配置虚拟机
spring.rabbitmq.virtual-host=/order
#采用手动应答
spring.rabbitmq.listener.simple.acknowledge-mode=manual
#开启发送确认机制
spring.rabbitmq.publisher-confirm-type=correlated
#可以确保消息在未被队列接收时返回
spring.rabbitmq.publisher-returns=true
##发送重试配置
#启用发送重试
#spring.rabbitmq.template.retry.enabled=true
#最大重试次数
#spring.rabbitmq.template.retry.max-attempts=5
#第一次和第二次尝试发布或传递消息之间的间隔
#spring.rabbitmq.template.retry.initial-interval=1000ms
#应用于上一重试间隔的乘数 步长
#spring.rabbitmq.template.retry.multiplier=2
#最大重试时间间隔
#spring.rabbitmq.template.retry.max-interval=10000ms
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

报错信息

rabbit PublisherCallbackChannel is closed问题跟踪

原因:

由于rabbit没有开启手动ack但是在消费监听中写了手动ack的代码导致信道关闭

解决：

spring.rabbitmq.listener.simple.acknowledge-mode=manual
1

生产者

  rabbitTemplate.convertAndSend("test",msg);
1

消费者

@Component
@Slf4j
public class Consumer {

    // 指定监听的队列名
    @RabbitListener(queues = "test")
    public void showMsg(Channel channel, Message message){
        String body=new String(message.getBody());
        System.out.println("收到的消息内容为：" + body);
        //确认收到消息
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            // redelivered = true, 表明该消息是重复处理消息
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            /**
             * 这里对消息重入队列做设置，例如将消息序列化缓存至 Redis, 并记录重入队列次数
             * 如果该消息重入队列次数达到一次次数，比如3次，将不再重入队列，直接拒绝
             * 这时候需要对消息做补偿机制处理
             *
             * channel.basicNack与channel.basicReject要结合越来使用
             *
             */
            try {
                if (redelivered) {

                    /**
                     * 1. 对于重复处理的队列消息做补偿机制处理
                     * 2. 从队列中移除该消息，防止队列阻塞
                     */
                    // 消息已重复处理失败, 扔掉消息
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
                    log.error("消息[{}]重新处理失败，扔掉消息", body);
                }

                // redelivered != true,表明该消息是第一次消费
                if (!redelivered) {

                    // 消息重新放回队列
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    log.error("消息[{}]处理失败，重新放回队列", body);
                }

            } catch (Exception e1) {
               log.error("showMsg",e1);
            }
        }
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

交换机

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hbVgkzY9-1662465913388)(C:\Users\Man\AppData\Roaming\Typora\typora-user-images\image-20220905192621453.png)]

直连交换机（Direct Exchange）

直连交换机，根据Routing Key(路由键)进行投递到不同队列。

单个绑定，一个路由键对应一个队列。如下所示：

多个绑定，一个路由键对应多个队列，则消息会分别投递到两个队列中，如下所示：

主题交换机（Topic Exchange）

主题交换机，对路由键进行模式匹配后进行投递

**#**表示可以匹配多个词

*****表示只能匹配一个词

示例：

路由*.*.key1 匹配如下：red.blue.key1、black.green.key1

路由#.key2.* 匹配如下：a.b.c.key2.red、orange.key2.blue、key2.red

路由key3.# 匹配如下：key3.red.blue、key3、key3.green、key3.black.blue.red

1
2
3
4
5
6

如下所示：

示例代码

1、topic配置文件

@Component
public class RabbitMqTopicConfig {

    private  static  String EXCHAN_NAME="/exchange_topic";
    private  static  String QUEUE_SMS="sms";
    private  static  String QUEUE_EMAIL="email";

    @Bean
    public Queue smsQueue(){
        return new Queue(QUEUE_SMS);
    }
    @Bean
    public Queue emailQueue(){
        return new Queue(QUEUE_EMAIL);
    }
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange(EXCHAN_NAME);
    }
    @Bean
    public Binding smsBinding(FanoutExchange fanoutExchange, Queue smsQueue){
        return BindingBuilder.bind(smsQueue).to(fanoutExchange);
    }
    @Bean
    public Binding emailBinding(FanoutExchange fanoutExchange, Queue emailQueue){
        return BindingBuilder.bind(emailQueue).to(fanoutExchange);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

2、对象

@Data
@AllArgsConstructor
public class User  implements Serializable {
    private String name;
    private int age;
    private  String sex;

}
1
2
3
4
5
6
7
8

3、生产者

@Autowired
    private RabbitTemplate rabbitTemplate;
    @GetMapping("/product")
    public  boolean sendMsg(@RequestParam(value = "name",required = false) String name){
        User user=new User(name,12,"男");
        rabbitTemplate.convertAndSend("/exchange_topic","test",user);
       return  true;
    }
1
2
3
4
5
6
7
8

4、消费者

@Component 
@Slf4j
public class Consumer {

    // 指定监听的队列名
    @RabbitListener(queues = "email")
    @RabbitHandler  // 消息接收处理
    public void showMsg(Channel channel, Message message, User user){
        System.out.println("收到的消息内容为：" + user);
        //确认收到消息
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            // redelivered = true, 表明该消息是重复处理消息
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            /**
             * 这里对消息重入队列做设置，例如将消息序列化缓存至 Redis, 并记录重入队列次数
             * 如果该消息重入队列次数达到一次次数，比如3次，将不再重入队列，直接拒绝
             * 这时候需要对消息做补偿机制处理
             *
             * channel.basicNack与channel.basicReject要结合越来使用
             *
             */
            try {
                if (redelivered) {

                    /**
                     * 1. 对于重复处理的队列消息做补偿机制处理
                     * 2. 从队列中移除该消息，防止队列阻塞
                     */
                    // 消息已重复处理失败, 扔掉消息
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
                    log.error("消息[{}]重新处理失败，扔掉消息",  user);
                }

                // redelivered != true,表明该消息是第一次消费
                if (!redelivered) {

                    // 消息重新放回队列
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    log.error("消息[{}]处理失败，重新放回队列", user);
                }

            } catch (Exception e1) {
               log.error("showMsg",e1);
            }
        }
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

扇形交换机（Fanout Exchange）

扇形交换机，采用广播模式，根据绑定的交换机，路由到与之对应的所有队列。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。

首部交换机（Headers Exchange）

头交换机，不处理路由键。而是根据发送的消息内容中的headers属性进行匹配。在绑定Queue与Exchange时指定一组键值对；当消息发送到RabbitMQ时会取到该消息的headers与Exchange绑定时指定的键值对进行匹配；如果完全匹配则消息会路由到该队列，否则不会路由到该队列。headers属性是一个键值对，可以是Hashtable，键值对的值可以是任何类型。而fanout，direct，topic 的路由键都需要要字符串形式的。

匹配规则x-match有下列两种类型：

x-match = all ：表示所有的键值对都匹配才能接受到消息

x-match = any ：表示只要有键值对匹配就能接受到消息

消息头交换机，如下图所示：

如何防止消息丢失

1、生产者没有成功把消息发送到MQ

丢失的原因：

因为网络传输的不稳定性，当生产者在向MQ发送消息的过程中，MQ没有成功接收到消息，但是生产者却以为MQ成功接收到了消息，不会再次重复发送该消息，从而导致消息的丢失。

解决办法：

有两个解决办法：事务机制和confirm机制，最常用的是confirm机制。

事务:

// 开启事务
 
channel.txSelect;
 
try {
 
  // 这里发送消息
 
} catch (Exception e) {
 
 channel.txRollback
 
 
 // 这里再次重发这条消息
 
}
 
// 提交事务
 
channel.txCommit;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

confirm机制：

RabbitMQ可以开启 confirm 模式，在生产者那里设置开启 confirm 模式之后，生产者每次写的消息都会分配一个唯一的 id，如果消息成功写入 RabbitMQ 中，RabbitMQ 会给生产者回传一个 ack 消息，告诉你说这个消息 ok 了。如果 RabbitMQ 没能处理这个消息，会回调你的一个 nack 接口，告诉你这个消息接收失败，生产者可以发送。而且你可以结合这个机制自己在内存里维护每个消息 id 的状态，如果超过一定时间还没接收到这个消息的回调，那么可以重发。

注意：

RabbitMQ的事务机制是同步的，很耗型能，会降低RabbitMQ的吞吐量。confirm机制是异步的，生成者发送完一个消息之后，不需要等待RabbitMQ的回调，就可以发送下一个消息，当RabbitMQ成功接收到消息之后会自动异步的回调生产者的一个接口返回成功与否的消息。

2、RabbitMQ接收到消息之后丢失了消息

丢失的原因：

RabbitMQ接收到生产者发送过来的消息，是存在内存中的，如果没有被消费完，此时RabbitMQ宕机了，那么再次启动的时候，原来内存中的那些消息都丢失了。

解决办法：

开启RabbitMQ的持久化。当生产者把消息成功写入RabbitMQ之后，RabbitMQ就把消息持久化到磁盘。结合上面的说到的confirm机制，只有当消息成功持久化磁盘之后，才会回调生产者的接口返回ack消息，否则都算失败，生产者会重新发送。存入磁盘的消息不会丢失，就算RabbitMQ挂掉了，重启之后，他会读取磁盘中的消息，不会导致消息的丢失。

持久化的配置：

• 第一点是创建 queue 的时候将其设置为持久化，这样就可以保证 RabbitMQ 持久化 queue 的元数据，但是它是不会持久化 queue 里的数据的。

   //声明队列   队列名、是否持久化、是否独立、是否自动删除、携带附加参数
              channel.queueDeclare(queue,true,false,false,null);
  1
  2

• 第二个是发送消息的时候将消息的 deliveryMode 设置为 2，就是将消息设置为持久化的，此时 RabbitMQ 就会将消息持久化到磁盘上去。

  // 交换机、路由key、deliveryMode 、消息
  channel.basicPublish("",queue, MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes(StandardCharsets.UTF_8));
        
  1
  2
  3

  注意：

  持久化要起作用必须同时设置这两个持久化才行，RabbitMQ 哪怕是挂了，再次重启，也会从磁盘上重启恢复 queue，恢复这个 queue 里的数据。

3、消费者弄丢了消息

丢失的原因：

如果RabbitMQ成功的把消息发送给了消费者，那么RabbitMQ的ack机制会自动的返回成功，表明发送消息成功，下次就不会发送这个消息。但如果就在此时，消费者还没处理完该消息，然后宕机了，那么这个消息就丢失了。

解决的办法：

简单来说，就是必须关闭 RabbitMQ 的自动 ack，可以通过一个 api 来调用就行，然后每次在自己代码里确保处理完的时候，再在程序里 ack 一把。这样的话，如果你还没处理完，不就没有 ack了？那 RabbitMQ 就认为你还没处理完，这个时候 RabbitMQ 会把这个消费分配给别的 consumer 去处理，消息是不会丢的。

消息应答机制

两种：自动应答、手动应答。

自动应答：

 /**
  * 消费者消费消息
  * 1.消费哪个队列
  * 2.消费成功之后是否要自动应答。 
  * true代表自动应答，false手动应答
  * 3.消费者，成功消费的回调
  * 4.消费者 取消消费的回调
*/
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);

1
2
3
4
5
6
7
8
9
10

手动应答:

//接收消息的回调
DeliverCallback deliverCallback = (consumerTag, delivery)-> {
            //休眠1s
            SleepUtils.sleep(1);
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Worker03-C1-接收到的消息："+ message);

//ACK肯定确认
//1.消息标记tag
//2.是否批量应答：false为不批量,true为批量
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
//取消消费的回调
CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println(consumerTag+ "：Worker03-C1-消息者 取消消费了-回调接口");
        };

//接收消息，采用手动应答
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

如何防止重复消费

重复消费原因：

正常情况下，消费者在消费消息的时候，消费完毕后，会发送一个确认消息给消息队列，消息队列就知道该消息被消费了，就会将该消息从消息队列中删除；但是因为网络传输等等故障，确认信息没有传送到消息队列，导致消息队列不知道自己已经消费过该消息了，再次将消息分发给其他的消费者。

解决思路：

保证消息的唯一性，就算是多次传输，不要让消息的多次消费带来影响；保证消息等幂性；

• 在消息生产时，MQ内部针对每条生产者发送的消息生成一个inner-msg-id，作为去重和幂等的依据（消息投递失败并重传），避免重复的消息进入队列；
• 在消息消费时，要求消息体中必须要有一个bizId（对于同一业务全局唯一，如支付ID、订单ID、帖子ID等）作为去重和幂等的依据，避免同一条消息被重复消费。

死信队列

没有被及时消费的消息存放的队列

ttl

ttl（time to live），消息存活时间

RabbitMQ支持两种ttl设置：

• 整个队列进行配置ttl，所有被投递到该队列的消息都最多不会存活超过N
• 单独消息进行配置ttl

如果同时配置了队列的TTL和消息的TTL，那么较小的那个值将会被使用。

死信消息

以下3种情况消息出现死信：

• 消费者拒收消息（basic.reject/ basic.nack） ，并且没有重新入队 requeue=false
• 消息在队列中未被消费，且超过队列或者消息本身的过期时间TTL(time-to-live)
• 队列的消息长度达到极限

出现死信时，此队列绑定了死信交换机，死信消息会路由到死信队列中

绑定死信队列

@Bean(BUCKET_TTL_QUEUE)
    public Queue bucketTtlQueue(){
        Map<String,Object> deadParamsMap = new HashMap<>();
        // 设置死信队列的Exchange
        deadParamsMap.put("x-dead-letter-exchange",BUCKET_DEAD_EXCHANGE);
        //设置死信队列的RouteKey
        deadParamsMap.put("x-dead-letter-routing-key",BUCKET_DEAD_QUEUE);
        // 设置对接过期时间"x-message-ttl"
        deadParamsMap.put("x-message-ttl",60000*5);//5分钟
        // 设置对接可以存储的最大消息数量
        //deadParamsMap.put("x-max-length",10);
        return new Queue(BUCKET_TTL_QUEUE,true,false,false,deadParamsMap);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14