RabbitMQ(一)之工作模式、生產(chǎn)者消息確認(rèn)機(jī)制

1、work queues 工作隊(duì)列模式

工作隊(duì)列模式是多個(gè)消費(fèi)者同時(shí)消費(fèi)同一個(gè)隊(duì)列中的數(shù)據(jù),默認(rèn)情況下,MQ會(huì)把所有請(qǐng)求平均發(fā)送給每一個(gè)消費(fèi)者??梢栽O(shè)置處理完成后在獲取下一條。channel.basicQos(1),消費(fèi)者偽代碼如下:

        //獲取TCP長(zhǎng)連接
        Connection conn = RabbitUtils.getConnection();
        //創(chuàng)建通信“通道”,相當(dāng)于TCP中的虛擬連接
        Channel channel = conn.createChannel();

        //創(chuàng)建隊(duì)列,聲明并創(chuàng)建一個(gè)隊(duì)列,如果隊(duì)列已存在,則使用這個(gè)隊(duì)列
        //第一個(gè)參數(shù):隊(duì)列名稱ID
        //第二個(gè)參數(shù):是否持久化,false對(duì)應(yīng)不持久化數(shù)據(jù),MQ停掉數(shù)據(jù)就會(huì)丟失
        //第三個(gè)參數(shù):是否隊(duì)列私有化,false則代表所有消費(fèi)者都可以訪問(wèn),true代表只有第一次擁有它的消費(fèi)者才能一直使用,其他消費(fèi)者不讓訪問(wèn)
        //第四個(gè):是否自動(dòng)刪除,false代表連接停掉后不自動(dòng)刪除掉這個(gè)隊(duì)列
        //其他額外的參數(shù), null
        channel.queueDeclare("rabbit-sms",false, false, false, null);
        //如果不寫(xiě)basicQos(1),則自動(dòng)MQ會(huì)將所有請(qǐng)求平均發(fā)送給所有消費(fèi)者
        //basicQos,MQ不再對(duì)消費(fèi)者一次發(fā)送多個(gè)請(qǐng)求,而是消費(fèi)者處理完一個(gè)消息后(確認(rèn)后),在從隊(duì)列中獲取一個(gè)新的
        channel.basicQos(1);//處理完一個(gè)取一個(gè)
        channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String jsonSMS = new String(body);
                System.out.println("短信發(fā)送成功:" + jsonSMS);
                //false只確認(rèn)簽收當(dāng)前的消息,設(shè)置為true的時(shí)候則代表簽收該消費(fèi)者所有未簽收的消息
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });

2、Pub、Sub 發(fā)布訂閱模式

后面的模式會(huì)使用到交換機(jī)(Exchange),交換機(jī)是接收生產(chǎn)者發(fā)送的消息,在根據(jù)Exchange類型不同區(qū)別發(fā)送到消費(fèi)者,或者將消息丟棄,Exchange只負(fù)責(zé)轉(zhuǎn)發(fā)消息,不具備存儲(chǔ)消息的能力。因此如果沒(méi)有任何隊(duì)列與Exchange綁定,或者沒(méi)有符合路由規(guī)則的隊(duì)列,那么消息會(huì)丟失

Exchange類型:

Fanout:廣播,將消息發(fā)送給所有綁定到交換機(jī)的隊(duì)列
Direct:定向,把消息發(fā)送給符合指定routingkey的隊(duì)列
Topic:通配符,把消息發(fā)送給符合routing pattern(路由模式)的隊(duì)列

發(fā)布訂閱模式是所有消費(fèi)者獲得的消息相同。與工作隊(duì)列模式不同的是,生產(chǎn)者和消費(fèi)者需要綁定到同一個(gè)交換機(jī)上。工作隊(duì)列模式不需要設(shè)置交換機(jī)(實(shí)際上,底層也會(huì)實(shí)用一個(gè)默認(rèn)的交換機(jī))。

生產(chǎn)者偽代碼:

        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.basicPublish("交換機(jī)名稱","" , null , "需要發(fā)送的內(nèi)容".getBytes());
        channel.close();
        connection.close();

消費(fèi)者偽代碼:

        Connection connection = RabbitUtils.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(RabbitConstant.QUEUE_BAIDU, false, false, false, null);

        //queueBind用于將隊(duì)列與交換機(jī)綁定
        channel.queueBind("自己的隊(duì)列名", "交換機(jī)名稱", "路由key(暫時(shí)用不到)");
        channel.basicQos(1);
        channel.basicConsume(RabbitConstant.QUEUE_BAIDU , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("訂閱到的信息:" + new String(body));
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });

3、Routing 路由模式

隊(duì)列與交換機(jī)的綁定需要制定一個(gè)RoutingKey(路由的key),消息發(fā)送方在向交換機(jī)(Exchange)發(fā)送消息時(shí),也必須指定RoutingKey,Exchange不在把消息發(fā)送給每一個(gè)綁定的隊(duì)列,而是根據(jù)消息的RoutingKey進(jìn)行判斷,只有綁定隊(duì)列的RoutingKey與生產(chǎn)消息的RoutingKey完全一致才會(huì)收到消息。RoutingKey可以綁定多個(gè)。

生產(chǎn)者偽代碼:

        //省略了創(chuàng)建連接和關(guān)閉連接的代碼
        channel.basicPublish("交換機(jī)的名稱","所要發(fā)送到哪個(gè)RoutingKey(RoutingKey1、RoutingKey2)" , null , "需要發(fā)送的內(nèi)容");

消費(fèi)者偽代碼:

        //省略了創(chuàng)建連接的代碼
        channel.queueDeclare("隊(duì)列名稱", false, false, false, null);
        //queueBind用于將隊(duì)列與交換機(jī)綁定
        //參數(shù)1:隊(duì)列名 參數(shù)2:交互機(jī)名  參數(shù)三:路由key
        channel.queueBind("隊(duì)列名稱", "Exchange  Name", "RoutingKey1");
        channel.queueBind("隊(duì)列名稱", "Exchange  Name", "RoutingKey2");
        channel.basicQos(1);
        channel.basicConsume("隊(duì)列名稱" , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接受到的信息:" + new String(body));
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });

4、Topic 通配符模式

Topic模式與Routing模式工作類似,Topic模式的Exchange可以讓隊(duì)列在綁定RoutingKey的時(shí)候使用通配符,RoutingKey一般是用一個(gè)或者多個(gè)單詞組成,例如:order.type1.pay、order.type2.pay。

通配符規(guī)則: #表示匹配一個(gè)或者多個(gè)詞,只能匹配一個(gè)詞。例如:order.#可以匹配上述兩個(gè)key,order.就不能匹配到,order.type1.*就可以匹配到order.type1.pay。

Topic模式下的發(fā)布者與Routing模式下的相同。只是消費(fèi)者有所區(qū)別

消費(fèi)者偽代碼如下:

//省略了創(chuàng)建鏈接的代碼
//RourtingKey可以使用通配符進(jìn)行匹配。order.#可以匹配到order.type1.pay、order.type2.pay,多個(gè)路由
channel.queueBind("隊(duì)列名稱", "Exchange Name", "order.#");

5、RabbitMQ的消息確認(rèn)機(jī)制

RabbitMQ提供了一個(gè)監(jiān)聽(tīng)器(Listener)來(lái)接收消息的投遞狀態(tài),消息確認(rèn)涉及到兩種狀態(tài),Confirm和Return:

Confirm代表生產(chǎn)者將消息送到Broker時(shí)產(chǎn)生的狀態(tài),包括:ack代表Broker已經(jīng)將數(shù)據(jù)接收成功,nack代表Broker拒收消息。原因可能有很多種,如隊(duì)列已滿、限流、IO異常等等。

Return代表消息已經(jīng)被Broker正常接收,消息已經(jīng)ack,但Broker沒(méi)有對(duì)應(yīng)的隊(duì)列進(jìn)行投遞時(shí),產(chǎn)生的狀態(tài),消息將被退回給生產(chǎn)者。

上述兩種狀態(tài)只代表生產(chǎn)者與Broker之間的消息投遞情況,與消費(fèi)者是否接收和確認(rèn)無(wú)關(guān)

        //省略了創(chuàng)建連接的代碼
        //開(kāi)啟confirm監(jiān)聽(tīng)模式
        channel.confirmSelect();
        //Confirm
        channel.addConfirmListener(new ConfirmListener() {
            public void handleAck(long l, boolean b) throws IOException {
                //第二個(gè)參數(shù)代表接收的數(shù)據(jù)是否為批量接收,一般我們用不到。
                System.out.println("消息已被Broker接收,Tag:" + l );
            }
            public void handleNack(long l, boolean b) throws IOException {
                System.out.println("消息已被Broker拒收,Tag:" + l);
            }
        });
        //Return
       channel.addReturnListener(new ReturnCallback() {
            public void handle(Return r) {
                System.err.println("===========================");
                System.err.println("Return編碼:" + r.getReplyCode() + "-Return描述:" + r.getReplyText());
                System.err.println("交換機(jī):" + r.getExchange() + "-路由key:" + r.getRoutingKey() );
                System.err.println("Return主題:" + new String(r.getBody()));
                System.err.println("===========================");
            }
        });
        //次處省略了發(fā)送消息的相關(guān)代碼
最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點(diǎn),簡(jiǎn)書(shū)系信息發(fā)布平臺(tái),僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容