欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

springboot整合RabbitMQ中死信隊(duì)列的實(shí)現(xiàn)

 更新時(shí)間：2024年10月23日 08:36:45   作者：饅頭警告  
死信是無(wú)法被消費(fèi)的消息,產(chǎn)生原因包括消息TTL過(guò)期、隊(duì)列最大長(zhǎng)度達(dá)到以及消息被拒絕且不重新排隊(duì),RabbitMQ的死信隊(duì)列機(jī)制能夠有效防止消息數(shù)據(jù)丟失,適用于訂單業(yè)務(wù)等場(chǎng)景,本文就來(lái)介紹一下

一、死信的概念

先從概念解釋上搞清楚這個(gè)定義，死信，顧名思義就是無(wú)法被消費(fèi)的消息，字面意思可以這樣理
解，一般來(lái)說(shuō)，producer 將消息投遞到 broker 或者直接到 queue 里了，consumer 從 queue 取出消息進(jìn)行消費(fèi)，但某些時(shí)候由于特定的原因?qū)е?queue 中的某些消息無(wú)法被消費(fèi)，這樣的消息如果沒(méi)有后續(xù)的處理，就變成了死信，有死信自然就有了死信隊(duì)列。

應(yīng)用場(chǎng)景:為了保證訂單業(yè)務(wù)的消息數(shù)據(jù)不丟失，需要使用到 RabbitMQ 的死信隊(duì)列機(jī)制，當(dāng)消息
消費(fèi)發(fā)生異常時(shí)，將消息投入死信隊(duì)列中.還有比如說(shuō): 用戶在商城下單成功并點(diǎn)擊去支付后在指定時(shí)間未支付時(shí)自動(dòng)失效

二、死信的來(lái)源

1、消息 TTL 過(guò)期
2、隊(duì)列達(dá)到最大長(zhǎng)度(隊(duì)列滿了，無(wú)法再添加數(shù)據(jù)到 mq 中)
3、消息被拒絕(basic.reject 或 basic.nack)并且 requeue=false.

三、死信實(shí)戰(zhàn)

1、代碼架構(gòu)圖

生產(chǎn)者正常情況下走的是普通的交換機(jī)，這個(gè)交換機(jī)的類型是 direct ，它和普通隊(duì)列之間的關(guān)系是一個(gè)叫 "zhangsan" 的路由 key， 正常情況下會(huì)被 C1 消費(fèi)。

但是發(fā)生了上面所說(shuō)的三種情況中的一種，成為了死信，然后被轉(zhuǎn)換到死信交換機(jī)中，這個(gè)死信交換機(jī)也是 direct 類型，它們之間的 routingKey 是 "lisi",然后就進(jìn)入了死信隊(duì)列，死信隊(duì)列由  C2 消費(fèi)。

2、消息 TTL 過(guò)期

（1）消費(fèi)者

// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 1
public class Comsumer01 {
    // 普通交換機(jī)名稱
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    // 死信交換機(jī)名稱
    public static final String DEAD_EXCHANGE = "dead_exchange";
    // 普通隊(duì)列名稱
    public static final String NORMAL_QUEUE = "normal_queue";
    // 死信隊(duì)列名稱
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = RabbitMqUtils.getChannel();
        // 交換機(jī)的聲明
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);

        // 普通隊(duì)列的聲明
        Map<String, Object> arguments = new HashMap<>();
        // 過(guò)期時(shí)間
        //arguments.put("x-message-ttl",100000);
        // 正常隊(duì)列設(shè)置死信交換機(jī)
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        // 設(shè)置死信 RoutingKey
        arguments.put("x-dead-letter-routing-key","lisi");
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
        
        
        // 死信隊(duì)列的聲明
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        // 綁定普通的交換機(jī)與普通隊(duì)列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        // 綁定死信的交換機(jī)與死信的隊(duì)列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        System.out.println("等待接收消息.....");

        DeliverCallback deliverCallback = ( consumerTag, message) ->{
            System.out.println("Consumer01 接收的消息是： " + new String(message.getBody(),"UTF-8"));
        };

        CancelCallback cancelCallback = consumerTag->{};

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);
    }
}
// 消費(fèi)者 2
public class Comsumer02 {
    // 死信隊(duì)列名稱
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("等待接收消息.....");

        DeliverCallback deliverCallback = ( consumerTag, message) ->{
            System.out.println("Consumer02 接收的消息是： " + new String(message.getBody(),"UTF-8"));
        };
        CancelCallback cancelCallback = consumerTag->{};
        
        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
    }
}

（2）生產(chǎn)者

// 死信隊(duì)列  生產(chǎn)者代碼
public class Producer {
    // 普通交換機(jī)名稱
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqUtils.getChannel();
        // 死信消息 設(shè)置 TTL 的時(shí)間
        AMQP.BasicProperties properties = new AMQP.BasicProperties().
                builder().expiration("10000").build();

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",properties,message.getBytes());

        }
    }
}

（3）結(jié)果展示

 3、隊(duì)列達(dá)到最大長(zhǎng)度

（1）消費(fèi)者

// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 1
public class Comsumer01 {
    // 普通交換機(jī)名稱
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    // 死信交換機(jī)名稱
    public static final String DEAD_EXCHANGE = "dead_exchange";
    // 普通隊(duì)列名稱
    public static final String NORMAL_QUEUE = "normal_queue";
    // 死信隊(duì)列名稱
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = RabbitMqUtils.getChannel();
        // 交換機(jī)的聲明
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);

        // 普通隊(duì)列的聲明
        Map<String, Object> arguments = new HashMap<>();
        // 過(guò)期時(shí)間
        //arguments.put("x-message-ttl",100000);
        // 正常隊(duì)列設(shè)置死信交換機(jī)
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        // 設(shè)置死信 RoutingKey
        arguments.put("x-dead-letter-routing-key","lisi");
        // 設(shè)置正常隊(duì)列的長(zhǎng)度的限制
        arguments.put("x-max-length",6);
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);


        // 死信隊(duì)列的聲明
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        // 綁定普通的交換機(jī)與普通隊(duì)列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        // 綁定死信的交換機(jī)與死信的隊(duì)列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        System.out.println("等待接收消息.....");

        DeliverCallback deliverCallback = ( consumerTag, message) ->{
            System.out.println("Consumer01 接收的消息是： " + new String(message.getBody(),"UTF-8"));
        };

        CancelCallback cancelCallback = consumerTag->{};

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);
    }
}
// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 2
public class Comsumer02 {
    // 死信隊(duì)列名稱
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("等待接收消息.....");

        DeliverCallback deliverCallback = ( consumerTag, message) ->{
            System.out.println("Consumer02 接收的消息是： " + new String(message.getBody(),"UTF-8"));
        };
        CancelCallback cancelCallback = consumerTag->{};

        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
    }
}

（2）生產(chǎn)者

// 死信隊(duì)列  生產(chǎn)者代碼
public class Producer {
    // 普通交換機(jī)名稱
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqUtils.getChannel();
        // 死信消息 設(shè)置 TTL 的時(shí)間
        /*AMQP.BasicProperties properties = new AMQP.BasicProperties().
                builder().expiration("10000").build();*/

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());

        }
    }
}

（3）結(jié)果展示

4、消息被拒

（1）消費(fèi)者

// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 1
public class Comsumer01 {
    // 普通交換機(jī)名稱
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    // 死信交換機(jī)名稱
    public static final String DEAD_EXCHANGE = "dead_exchange";
    // 普通隊(duì)列名稱
    public static final String NORMAL_QUEUE = "normal_queue";
    // 死信隊(duì)列名稱
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = RabbitMqUtils.getChannel();
        // 交換機(jī)的聲明
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);

        // 普通隊(duì)列的聲明
        Map<String, Object> arguments = new HashMap<>();
        // 過(guò)期時(shí)間
        //arguments.put("x-message-ttl",100000);
        // 正常隊(duì)列設(shè)置死信交換機(jī)
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        // 設(shè)置死信 RoutingKey
        arguments.put("x-dead-letter-routing-key","lisi");
        // 設(shè)置正常隊(duì)列的長(zhǎng)度的限制
        // arguments.put("x-max-length",6);
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);


        // 死信隊(duì)列的聲明
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        // 綁定普通的交換機(jī)與普通隊(duì)列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        // 綁定死信的交換機(jī)與死信的隊(duì)列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        System.out.println("等待接收消息.....");

        DeliverCallback deliverCallback = ( consumerTag, message) ->{
            String msg = new String(message.getBody(),"UTF-8");
            if (msg.equals("info5")){
                System.out.println("Consumer01 接收的消息是： " + msg + ": 此消息是被 C1 拒絕的");
                // 拒絕，且不放囧普通隊(duì)列
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }else {
                System.out.println("Consumer01 接收的消息是： " + new String(message.getBody(),"UTF-8"));   
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);            
            }
        };

        CancelCallback cancelCallback = consumerTag->{};

        // 開啟手動(dòng)應(yīng)答
        channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,cancelCallback);
    }
}
// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 2
public class Comsumer02 {
    // 死信隊(duì)列名稱
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("等待接收消息.....");

        DeliverCallback deliverCallback = ( consumerTag, message) ->{
            System.out.println("Consumer02 接收的消息是： " + new String(message.getBody(),"UTF-8"));
        };
        CancelCallback cancelCallback = consumerTag->{};

        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
    }
}

（2）生產(chǎn)者

// 死信隊(duì)列  生產(chǎn)者代碼
public class Producer {
    // 普通交換機(jī)名稱
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqUtils.getChannel();
        // 死信消息 設(shè)置 TTL 的時(shí)間
        /*AMQP.BasicProperties properties = new AMQP.BasicProperties().
                builder().expiration("10000").build();*/

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());

        }
    }
}

（3）結(jié)果展示

到此這篇關(guān)于springboot整合RabbitMQ中死信隊(duì)列的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)springboot RabbitMQ死信隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

相關(guān)文章

  • SpringBoot+websocket實(shí)現(xiàn)消息對(duì)話功能

    SpringBoot+websocket實(shí)現(xiàn)消息對(duì)話功能

    WebSocket是一種在Web應(yīng)用程序中實(shí)現(xiàn)實(shí)時(shí)雙向通信的技術(shù),它可以用于在線游戲、在線聊天、推送通知、實(shí)時(shí)監(jiān)控等,并且比傳統(tǒng)的輪詢技術(shù)更加高效和可靠,本文就給大家介紹基于SpringBoot+websocket實(shí)現(xiàn)消息對(duì)話功能,感興趣的小伙伴可以自己動(dòng)手試一試
    2023-09-09
  • SpringBoot整合Log4j2及配置步驟

    SpringBoot整合Log4j2及配置步驟

    這篇文章主要介紹了SpringBoot整合Log4j2以及配置詳解,刪除spring-boot-starter-parent默認(rèn)使用spring-boot-starter-logging依賴，本文結(jié)合實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下
    2023-01-01
  • Spring Security基本架構(gòu)與初始化操作流程詳解

    Spring Security基本架構(gòu)與初始化操作流程詳解

    這篇文章主要介紹了Spring Security基本架構(gòu)與初始化操作流程，Spring Security是一個(gè)能夠?yàn)榛赟pring的企業(yè)應(yīng)用系統(tǒng)提供聲明式的安全訪問(wèn)控制解決方案的安全框架
    2023-03-03
  • JWT登錄認(rèn)證Springboot詳解

    JWT登錄認(rèn)證Springboot詳解

    文章主要介紹了如何在Java項(xiàng)目中使用JWT進(jìn)行用戶認(rèn)證和授權(quán),通過(guò)定義一個(gè)常量,編寫JWT工具類來(lái)生成和解析token,登錄時(shí)在服務(wù)端生成token并返回給客戶端,客戶端使用攔截器攔截請(qǐng)求,驗(yàn)證token的有效性,從而實(shí)現(xiàn)權(quán)限控制,文章旨在分享個(gè)人經(jīng)驗(yàn),為開發(fā)者提供參考
    2024-11-11
  • Comet框架Plushlet使用問(wèn)題解決方案

    Comet框架Plushlet使用問(wèn)題解決方案

    這篇文章主要介紹了Comet框架Plushlet使用問(wèn)題解決方案,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下
    2020-10-10
  • Elasticsearch倒排索引詳解及實(shí)際應(yīng)用中的優(yōu)化

    Elasticsearch倒排索引詳解及實(shí)際應(yīng)用中的優(yōu)化

    Elasticsearch(ES)使用倒排索引來(lái)加速文本的搜索速度,倒排索引之所以高效,主要是因?yàn)樗淖兞藬?shù)據(jù)的組織方式,使得查詢操作可以快速完成,這篇文章主要給大家介紹了關(guān)于Elasticsearch倒排索引詳解及實(shí)際應(yīng)用中優(yōu)化的相關(guān)資料,需要的朋友可以參考下
    2024-08-08
  • Java中Word與PDF轉(zhuǎn)換為圖片的方法詳解

    Java中Word與PDF轉(zhuǎn)換為圖片的方法詳解

    這篇文章主要為大家詳細(xì)介紹了如何使用Java實(shí)現(xiàn)將Word與PDF轉(zhuǎn)換為圖片,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下
    2024-10-10
  • Java精確抽取網(wǎng)頁(yè)發(fā)布時(shí)間

    Java精確抽取網(wǎng)頁(yè)發(fā)布時(shí)間

    這篇文章主要為大家詳細(xì)介紹了Java精確抽取網(wǎng)頁(yè)發(fā)布時(shí)間的相關(guān)資料，盡量做到精確無(wú)誤，感興趣的小伙伴們可以參考一下
    2016-06-06
  • Java使用Redis實(shí)現(xiàn)微博熱搜功能

    Java使用Redis實(shí)現(xiàn)微博熱搜功能

    在社交平臺(tái)上,熱搜功能是一個(gè)非常重要的組成部分,它展示了當(dāng)前最熱門的話題,幫助用戶迅速了解最受關(guān)注的事件,Redis 是一個(gè)高性能的鍵值存儲(chǔ)系統(tǒng),通常用于緩存和實(shí)時(shí)數(shù)據(jù)存儲(chǔ),本文將通過(guò) Java 結(jié)合 Redis 實(shí)現(xiàn)一個(gè)簡(jiǎn)化版的微博熱搜功能,需要的朋友可以參考下
    2024-12-12
  • SpringBoot接受前臺(tái)參數(shù)的6種方式以及統(tǒng)一響應(yīng)代碼示例

    SpringBoot接受前臺(tái)參數(shù)的6種方式以及統(tǒng)一響應(yīng)代碼示例

    這篇文章主要給大家介紹了關(guān)于SpringBoot接受前臺(tái)參數(shù)的6種方式以及統(tǒng)一響應(yīng)的相關(guān)資料,前端負(fù)責(zé)展示頁(yè)面和用戶交互,而后端則負(fù)責(zé)處理業(yè)務(wù)邏輯和數(shù)據(jù)存儲(chǔ),在這種架構(gòu)下前端需要將用戶輸入的數(shù)據(jù)發(fā)送給后端進(jìn)行處理,需要的朋友可以參考下
    2023-12-12

最新評(píng)論