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

Java消息隊(duì)列RabbitMQ之消息回調(diào)詳解

 更新時(shí)間：2023年07月31日 09:43:39   作者：迷鹿小女子  
這篇文章主要介紹了Java消息隊(duì)列RabbitMQ之消息回調(diào)詳解,消息回調(diào)，其實(shí)就是消息確認(rèn)(生產(chǎn)者推送消息成功，消費(fèi)者接收消息成功)  , 對于程序來說，發(fā)送者沒法確認(rèn)是否發(fā)送成功,需要的朋友可以參考下

消息100%的投遞

消息如何保障100%的投遞成功？ 什么是生產(chǎn)端的可靠性投遞？

  • 保障消息的成功發(fā)出
  • 保障MQ節(jié)點(diǎn)的成功接收
  • 發(fā)送端收到MQ節(jié)點(diǎn)Broker確認(rèn)應(yīng)答
  • 完善的消息進(jìn)行補(bǔ)償機(jī)制

BAT/TMD互聯(lián)網(wǎng)大廠的解決方案：

  • 消息落庫，對消息狀態(tài)進(jìn)行打標(biāo)
  • 消息的延遲投遞，做二次確認(rèn)，回調(diào)檢查

在這里插入圖片描述

在這里插入圖片描述

冪等性概念

冪等性是什么？

  • 我們可以借鑒數(shù)據(jù)庫的樂觀鎖機(jī)制
  • 比如我們執(zhí)行一條更新庫存的SQL語句
  • pdate t_repository set cont = cont -1,version = version + 1 where version = 1
  • Elasticsearch也是嚴(yán)格遵循冪等性概念，每次數(shù)據(jù)更新，version+1

消費(fèi)端-冪等性保障

在海量訂單產(chǎn)生的業(yè)務(wù)高峰期，如何避免消息的重復(fù)消費(fèi)問題？

消費(fèi)實(shí)現(xiàn)冪等性，就意味著，我們的消息永遠(yuǎn)不會消費(fèi)多次，即使我們收到了多條一樣的消息

業(yè)界主流的冪等性操作

唯一ID+指紋碼機(jī)制，利用數(shù)據(jù)庫主鍵去重 利用Redis的原子性去實(shí)現(xiàn)

  • 唯一ID+指紋碼機(jī)制，利用數(shù)據(jù)庫主鍵去重
  • Select count(1) from T_order where ID=唯一ID+指紋碼

好處：實(shí)現(xiàn)簡單

壞處：高并發(fā)下有數(shù)據(jù)庫寫入的性能瓶頸

解決方案：根據(jù)ID進(jìn)行分庫分表進(jìn)行算法路由

利用Redis的原子性去實(shí)現(xiàn)

使用Redis進(jìn)行冪等，需要考慮的問題

第一：我們是否要進(jìn)行數(shù)據(jù)落庫，如果落庫的話，關(guān)鍵解決的問題是數(shù)據(jù)庫和緩存如何做到原子性？

第二：如果不進(jìn)行落庫，那么都存儲到緩存中，如何設(shè)置定時(shí)同步策略？

Confirm確認(rèn)消息

理解Confirm消息確認(rèn)機(jī)制 消息的確認(rèn)，是指生產(chǎn)者投遞消息后，如果Broker收到消息，則會給我們生產(chǎn)者 一個(gè)應(yīng)答。

生產(chǎn)者進(jìn)行接收應(yīng)答，用來確定這條消息是否正常的發(fā)送到Broker，這種方式也是消息的可靠性投遞的核心保障

在這里插入圖片描述

如何實(shí)現(xiàn)Confirm確認(rèn)消息？

第一步：在Channel上開啟確認(rèn)模式：channel.confirmSelect()

第二步：在channel上添加監(jiān)聽：addConfirmListener，監(jiān)聽成功和失敗的返回結(jié)果，根據(jù)具體的結(jié)果對消息進(jìn)行重新發(fā)送、或記錄日志等后續(xù)處理！

消費(fèi)端代碼

package com.xieminglu.rabbitmqapi.confirm;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Consumer {
    public static void main(String[] args) throws Exception {
        //1 創(chuàng)建ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.248.132");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        //2 獲取C    onnection
        Connection connection = connectionFactory.newConnection();
        //3 通過Connection創(chuàng)建一個(gè)新的Channel
        Channel channel = connection.createChannel();
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.#";
        String queueName = "test_confirm_queue";
        //4 聲明交換機(jī)和隊(duì)列 然后進(jìn)行綁定設(shè)置, 最后制定路由Key
        channel.exchangeDeclare(exchangeName, "topic", true);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        //5 創(chuàng)建消費(fèi)者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        channel.basicConsume(queueName, true, queueingConsumer);
        while(true){
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.err.println("消費(fèi)端: " + msg);
        }
    }
}

服務(wù)提供方代碼

package com.xieminglu.rabbitmqapi.confirm;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
public class Producer {
    public static void main(String[] args) throws Exception {
        //1 創(chuàng)建ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.248.132");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        //2 獲取C    onnection
        Connection connection = connectionFactory.newConnection();
        //3 通過Connection創(chuàng)建一個(gè)新的Channel
        Channel channel = connection.createChannel();
        //4 指定我們的消息投遞模式: 消息的確認(rèn)模式
        channel.confirmSelect();
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.save";
        //5 發(fā)送一條消息
        String msg = "Hello RabbitMQ Send confirm message!";
        channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
        //6 添加一個(gè)確認(rèn)監(jiān)聽
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("-------no ack!-----------");
            }
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("-------ack!-----------");
            }
        });
    }
}

Retrn返回消息

Retrn Listener用于處理一些不可路由的消息！

  • 正常情況：我們的消息生產(chǎn)者，通過指定一個(gè)Exchange和RotingKey，把消息送達(dá)到某一個(gè)隊(duì)列中去，然后我們的消費(fèi)者監(jiān)聽隊(duì)列，進(jìn)行消費(fèi)處理操作！
  • 異常情況：在某些情況下，如果我們在發(fā)送消息的時(shí)候，當(dāng)前的Exchange不存在或者指定的路由key路由不到，這個(gè)時(shí)候如果我們需要監(jiān)聽這種不可達(dá)的消息，就需要使用Retrn Listener！

在基礎(chǔ)API中有一個(gè)關(guān)鍵的配置項(xiàng) Mandatory：如果為tre，則監(jiān)聽器會接收到路由不可達(dá)的消息，然后進(jìn)行后續(xù)處理，如果為false，那么Broker端自動刪除該消息！

在這里插入圖片描述

消費(fèi)端代碼

package com.xieminglu.rabbitmqapi.returnlistener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Consumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.248.132");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        String exchangeName = "test_return_exchange";
        String routingKey = "return.#";
        String queueName = "test_return_queue";
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        channel.basicConsume(queueName, true, queueingConsumer);
        while(true){
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.err.println("消費(fèi)者: " + msg);
        }
    }
}

生產(chǎn)端代碼

package com.xieminglu.rabbitmqapi.returnlistener;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Producer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.248.132");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        String exchange = "test_return_exchange";
        String routingKey = "return.save";
        String routingKeyError = "abc.save";
        String msg = "Hello RabbitMQ Return Message";
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange,
                                     String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.err.println("---------handle  return----------");
                System.err.println("replyCode: " + replyCode);
                System.err.println("replyText: " + replyText);
                System.err.println("exchange: " + exchange);
                System.err.println("routingKey: " + routingKey);
                System.err.println("properties: " + properties);
                System.err.println("body: " + new String(body));
            }
        });
        //消息投遞成功，會被消費(fèi)者所消費(fèi)
        channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
        //消息不可達(dá)，將觸發(fā)ReturnListener
//         channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
    }
}

自定義消費(fèi)者

我們一般就是在代碼中編寫while循環(huán)，進(jìn)行consmer.nextDelivery方法進(jìn)行獲取下一條消息，然后進(jìn)行消費(fèi)處理！

但是我們使用自定義的Consmer更加的方便，解耦性更加的強(qiáng)，也是實(shí)際工作中最常用的使用方式！

自定義消費(fèi)端代碼

package com.xieminglu.rabbitmqapi.consumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
public class MyConsumer extends DefaultConsumer {
    public MyConsumer(Channel channel) {
        super(channel);
    }
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        System.err.println("consumerTag: " + consumerTag);
        System.err.println("envelope: " + envelope);
        System.err.println("properties: " + properties);
        System.err.println("body: " + new String(body));
    }
}

消費(fèi)端調(diào)用

package com.xieminglu.rabbitmqapi.consumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Consumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.248.132");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        String exchangeName = "test_consumer_exchange";
        String routingKey = "consumer.#";
        String queueName = "test_consumer_queue";
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        channel.basicConsume(queueName, true, new MyConsumer(channel));
        }
        }

生產(chǎn)端調(diào)用

package com.xieminglu.rabbitmqapi.consumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.248.132");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        String exchange = "test_consumer_exchange";
        String routingKey = "consumer.save";
        String msg = "Hello RabbitMQ Consumer Message";
        for(int i =0; i<5; i ++){
            channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
        }
    }
}

到此這篇關(guān)于Java消息隊(duì)列RabbitMQ之消息回調(diào)詳解的文章就介紹到這了,更多相關(guān)RabbitMQ消息回調(diào)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

相關(guān)文章

  • Java跨模塊調(diào)用方式

    Java跨模塊調(diào)用方式

    這篇文章主要介紹了Java跨模塊調(diào)用方式,具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教
    2024-08-08
  • 三道java新手入門面試題,通往自由的道路--多線程

    三道java新手入門面試題,通往自由的道路--多線程

    這篇文章主要為大家分享了最有價(jià)值的3道多線程面試題，涵蓋內(nèi)容全面，包括數(shù)據(jù)結(jié)構(gòu)和算法相關(guān)的題目、經(jīng)典面試編程題等，對hashCode方法的設(shè)計(jì)、垃圾收集的堆和代進(jìn)行剖析，感興趣的小伙伴們可以參考一下
    2021-07-07
  • Spring?Boot存在路徑遍歷漏洞CVE-2021-22118的問題解析

    Spring?Boot存在路徑遍歷漏洞CVE-2021-22118的問題解析

    CVE-2021-22118?是一個(gè)在?Spring?Boot?中發(fā)現(xiàn)的漏洞，該漏洞關(guān)系到?Spring?Boot?的開發(fā)者工具（Devtools）中的遠(yuǎn)程更新（Remote?Update）功能，這篇文章主要介紹了Spring?Boot存在路徑遍歷漏洞CVE-2021-22118,需要的朋友可以參考下
    2023-09-09
  • springboot與vue詳解實(shí)現(xiàn)短信發(fā)送流程

    springboot與vue詳解實(shí)現(xiàn)短信發(fā)送流程

    隨著人工智能的不斷發(fā)展，機(jī)器學(xué)習(xí)這門技術(shù)也越來越重要，很多人都開啟了學(xué)習(xí)機(jī)器學(xué)習(xí)，本文就介紹了機(jī)器學(xué)習(xí)的基礎(chǔ)內(nèi)容
    2022-06-06
  • Spring boot AOP通過XML配置文件聲明的方法

    Spring boot AOP通過XML配置文件聲明的方法

    這篇文章主要介紹了Spring boot AOP通過XML配置文件聲明，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2020-06-06
  • Java的JDBC中Statement與CallableStatement對象實(shí)例

    Java的JDBC中Statement與CallableStatement對象實(shí)例

    這篇文章主要介紹了Java的JDBC中Statement與CallableStatement對象實(shí)例,JDBC是Java編程中用于操作數(shù)據(jù)庫的API,需要的朋友可以參考下
    2015-12-12
  • Java虛擬機(jī)類加載器之雙親委派機(jī)制模型案例

    Java虛擬機(jī)類加載器之雙親委派機(jī)制模型案例

    這篇文章主要介紹了Java虛擬機(jī)類加載器之雙親委派機(jī)制模型案例,本篇文章通過簡要的案例,講解了該項(xiàng)技術(shù)的了解與使用,以下就是詳細(xì)內(nèi)容,需要的朋友可以參考下
    2021-08-08
  • Java實(shí)現(xiàn)圖片合成的示例詳解

    Java實(shí)現(xiàn)圖片合成的示例詳解

    前端有一個(gè)神器——canvas，這個(gè)畫布標(biāo)簽可以處理各種圖片的合成，可以精確到圖片的具體坐標(biāo)。java后端也有這樣的神器，那就是image-combiner，可以很簡單的合成圖片，感興趣的可以試一試
    2022-01-01
  • Go Java算法之簡化路徑實(shí)例詳解

    Go Java算法之簡化路徑實(shí)例詳解

    這篇文章主要為大家介紹了Go Java算法之簡化路徑實(shí)例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪
    2022-09-09
  • Java語言實(shí)現(xiàn)快速冪取模算法詳解

    Java語言實(shí)現(xiàn)快速冪取模算法詳解

    這篇文章主要介紹了Java語言實(shí)現(xiàn)快速冪取模算法詳解，具有一定參考價(jià)值，需要的朋友可以了解下。
    2017-11-11

最新評論