快捷導(dǎo)航

SpringBoot整合RabbitMQ實(shí)現(xiàn)消息確認(rèn)機(jī)制

更新時(shí)間：2021年08月31日 14:10:17 作者：過期可樂

這篇文章主要介紹了SpringBoot整合RabbitMQ實(shí)現(xiàn)消息確認(rèn)機(jī)制,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

前面幾篇案例已經(jīng)將常用的交換器（DirectExchange、TopicExchange、FanoutExchange）的用法介紹完了，現(xiàn)在我們來看一下消息的回調(diào)，也就是消息確認(rèn)。

在rabbitmq-provider項(xiàng)目的application.yml文件上加上一些配置

server:
  port: 8021
spring:
  #給項(xiàng)目來個(gè)名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服務(wù)器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
#    connection-timeout: 60s
    #虛擬host 可以不設(shè)置,使用server默認(rèn)host
    #virtual-host: admin
    # 確認(rèn)消息已發(fā)送到交換機(jī)（Exchange）
    publisher-confirm-type: correlated
    # 確認(rèn)消息已發(fā)送到隊(duì)列
    publisher-returns: true

創(chuàng)建RabbitConfig配置類，配置消息的相關(guān)回調(diào)函數(shù)

package com.rabbitmq.provider.config;

import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class RabbitConfig {

    @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        // 設(shè)置開啟Mandatory，才能觸發(fā)回調(diào)函數(shù)，無論消息推送結(jié)果怎么樣都強(qiáng)制調(diào)用回調(diào)函數(shù)
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                System.out.println("ConfirmCallback    ：相關(guān)數(shù)據(jù)：" + correlationData);
                System.out.println("ConfirmCallback    ：確認(rèn)情況：" + b);
                System.out.println("ConfirmCallback    ：原因：" + s);
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("ReturnCallback    ：消息：" + message);
                System.out.println("ReturnCallback    ：回應(yīng)碼：" + i);
                System.out.println("ReturnCallback    ：回應(yīng)消息：" + s);
                System.out.println("ReturnCallback    ：交換機(jī)：" + s1);
                System.out.println("ReturnCallback    ：路由鍵：" + s2);
            }
        });
        return rabbitTemplate;
    }
}

生產(chǎn)者回調(diào)函數(shù)已經(jīng)配置完畢，上面我們配置了兩個(gè)回調(diào)函數(shù)setConfirmCallback和setReturnCallback

一般會(huì)產(chǎn)生四種情況：

①消息推送到server，但是在server里找不到交換機(jī)
②消息推送到server，找到交換機(jī)了，但是沒找到隊(duì)列
③消息推送到sever，交換機(jī)和隊(duì)列啥都沒找到
④消息推送成功

接下來分別測(cè)試和認(rèn)證下以上4種情況，消息確認(rèn)觸發(fā)回調(diào)函數(shù)的情況：

①消息推送到server，但是在server里找不到交換機(jī)

寫個(gè)測(cè)試接口，把消息推送到名為‘non-existent-exchange'的交換機(jī)上（這個(gè)交換機(jī)是沒有創(chuàng)建沒有配置的）：

@GetMapping("/testMessageAck")
    public String testMessageAck() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
        return "ok";
    }

調(diào)用接口，查看rabbitmq-provuder項(xiàng)目的控制臺(tái)輸出情況（原因里面有說，沒有找到交換機(jī)'non-existent-exchange'）：

所以證明在交換器沒有配置，推送消息找不到對(duì)應(yīng)的交換器，會(huì)執(zhí)行ConfirmCallback方法。

②消息推送到server，找到交換機(jī)了，但是沒找到隊(duì)列

我們?cè)赗abbitConfig里面配置了一個(gè)名為lonelyDirectExchange的交換器，再新增一個(gè)接口testMessageAck2，使用lonelyDirectExchange交換器

@GetMapping("/testMessageAck2")
    public String testMessageAck2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
        return "ok";
    }

重啟服務(wù)，使用postman調(diào)用這個(gè)接口

可以看到ReturnCallback和ConfirmCallback這兩個(gè)都被調(diào)用了。

CallfirmCallback的確認(rèn)情況為true，因?yàn)橄⑹峭扑统晒Φ椒?wù)器了的，所以ConfirmCallback對(duì)消息確認(rèn)情況是true

ReturnCallback的回應(yīng)消息為NO_ROUTE找不到隊(duì)列

③消息推送到sever，交換機(jī)和隊(duì)列啥都沒找到

這種情況和1、2很相似，會(huì)調(diào)用ReturnCallback和CallfirmCallback這兩個(gè)回調(diào)函數(shù)

④消息推送成功

我們調(diào)用之前的sendFanoutMessage接口來看一下會(huì)是什么情況

結(jié)論：消息推送成功調(diào)用ConfirmCallback回調(diào)函數(shù)

消息確認(rèn)機(jī)制

消息接收確認(rèn)機(jī)制主要存在三種模式：

①：自動(dòng)確認(rèn)

②：根據(jù)情況確認(rèn)

③：手動(dòng)確認(rèn)

接下來，我們看看手動(dòng)確認(rèn)是怎么實(shí)現(xiàn)的

在rabbitmq-consumer消費(fèi)者項(xiàng)目上創(chuàng)建MessageListenerConfig配置類

package com.rabbitmq.consumer.config;

import com.rabbitmq.consumer.receiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class MessageListenerConfig {

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Autowired
    private MyAckReceiver myAckReceiver;    //消息接收處理類

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        // RabbitMQ 默認(rèn)是自動(dòng)確認(rèn)，改成手動(dòng)確認(rèn)
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 設(shè)置隊(duì)列
        // 如果同時(shí)設(shè)置多個(gè)如下:前提是隊(duì)列是必須已經(jīng)創(chuàng)建存在的
        // container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
        // 設(shè)置一個(gè)隊(duì)列
        // container.setQueues(new Queue("TestDirectQueue",true));
        // 設(shè)置多個(gè)隊(duì)列
         container.addQueues(new Queue("TestDirectQueue",true));
         container.addQueues(new Queue("fanout.A",true));
        // container.addQueues(new Queue("TestDirectQueue3",true));

        container.setMessageListener(myAckReceiver);
        return container;
    }

}

創(chuàng)建手動(dòng)確定消息監(jiān)聽類MyAckReceiver

package com.rabbitmq.consumer.receiver;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;


@Component
public class MyAckReceiver implements ChannelAwareMessageListener {


    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因?yàn)閭鬟f消息的時(shí)候用的map傳遞，所以將map從Message內(nèi)取出來需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");
            Map<String,String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId = msgMap.get("messageId");
            String messageData = msgMap.get("messageData");
            String createTime = msgMap.get("createTime");
            System.out.println("MyAckReceiver ：");
            System.out.println("messageId = " + messageId);
            System.out.println("messageData = " + messageData);
            System.out.println("createTime = " + createTime);
            System.out.println("消費(fèi)的主題來自：" + message.getMessageProperties().getConsumerQueue());
            if("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("執(zhí)行TestDirectQueue中的消息的業(yè)務(wù)處理流程.....");
            }
            if("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("執(zhí)行fanout.A中的消息的業(yè)務(wù)處理流程.....");
            }
            //第二個(gè)參數(shù)，手動(dòng)確認(rèn)可以被批處理，當(dāng)該參數(shù)為 true 時(shí)，則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息
            channel.basicAck(deliveryTag,true);
            //第二個(gè)參數(shù)，true會(huì)重新放回隊(duì)列，所以需要自己根據(jù)業(yè)務(wù)邏輯判斷什么時(shí)候使用拒絕
            //channel.basicReject(deliveryTag, true);
        }catch (Exception e){
            channel.basicAck(deliveryTag,false);
            e.printStackTrace();
        }
    }
    // {key=value,key=value,key=value} 格式轉(zhuǎn)換成map
    private Map<String,String> mapStringToMap(String str,int entryNum) {
        str = str.substring(1, str.length()-1);
        String[] strs = str.split(",", entryNum);
        Map<String,String> map = new HashMap<String,String>();
        for (String s : strs) {
            String key = s.split("=")[0].trim();
            String value = s.split("=")[1];
            map.put(key,value);
        }
        return map;
    }
}

重啟服務(wù)，使用postman調(diào)用sendDirectMessage接口

再調(diào)用sendFanoutMessage接口

如果還想添加其它處處理器手動(dòng)確認(rèn)，按照上面的模式繼續(xù)添加即可。

到此這篇關(guān)于SpringBoot整合RabbitMQ之消息確認(rèn)的文章就介紹到這了,更多相關(guān)SpringBoot整合RabbitMQ消息確認(rèn)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: