2. 程式人生 > >RabbitMQ使用場景練習：訊息確認機制(十一)

RabbitMQ使用場景練習：訊息確認機制(十一)

阿新 • • 發佈：2019-01-03
  • 訊息確認機制
RabbitMQ提供了transaction、confirm兩種訊息確認機制。transaction即事務機制，手動提交和回滾；confirm機制提供了Confirmlistener和waitForConfirms兩種方式。confirm機制效率明顯會高於transaction機制，但後者的優勢在於強一致性。如果沒有特別的要求，建議使用conrim機制。 

1、從實驗來看，訊息的確認機制只是確認publisher傳送訊息到broker，由broker進行應答，不能確認訊息是否有效消費。 
2、而為了確認訊息是否被髮送給queue，應該在傳送訊息中啟用引數mandatory=true，使用ReturnListener接收未被髮送成功的訊息。 

3、接下來就需要確認訊息是否被有效消費。publisher端目前並沒有提供監聽事件，但提供了應答機制來保證訊息被成功消費，應答方式： 
   basicAck：成功消費，訊息從佇列中刪除 
   basicNack：requeue=true，訊息重新進入佇列，false被刪除 
   basicReject：等同於basicNack 
   basicRecover：訊息重入佇列，requeue=true，傳送給新的consumer，false傳送給相同的consumer 

  • 應答模式之transaction機制

需要手動提交和回滾，執行txCommit，訊息才會轉發給佇列進入ready狀態；執行txRollback，訊息被取消 

package com.demo.mq.rabbitmq.example11;  
import java.io.IOException;  
import java.io.Serializable;  
import org.apache.commons.lang3.SerializationUtils;  
import com.demo.mq.rabbitmq.MqManager;  
import com.rabbitmq.client.AMQP;  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.DefaultConsumer;  
import com.rabbitmq.client.Envelope;  
  
/** 
 * 應答模式之transaction機制 
 * @author sheungxin 
 * 
 */  
public class TxDemo {  
    private static String exchange_name="";  
    private static String queue_name="tx_queue";  
      
    /** 
     * transaction機制傳送訊息,事務機制：手動提交和回滾 
     * 執行txCommit，訊息才會轉發給佇列進入ready狀態 
     * 執行txRollback，訊息被取消 
     * @param mes 
     * @throws Exception 
     */  
    public static void txSend(Serializable mes) throws Exception{  
        Connection conn=MqManager.newConnection();  
        Channel channel=conn.createChannel();  
        //開啟transaction機制  
        channel.txSelect();  
        channel.queueDeclare(queue_name,false,false,true,null);  
        for(int i=0;i<10;i++){  
            try{  
                channel.basicPublish(exchange_name, queue_name, null, SerializationUtils.serialize(mes.toString()+i));  
                //do something  
//              int n=5/0;//試驗訊息回滾  
                channel.txCommit();//提交訊息  
                System.out.println("釋出訊息"+mes.toString()+i);  
            }catch(Exception e){  
                channel.txRollback();//異常，取消訊息  
                System.out.println("回滾訊息"+mes.toString()+i);  
            }  
        }  
    }  
  
    /** 
     * transaction機制接收訊息,事務機制：手動提交和回滾 
     * 消費者需要執行basicAck，並txCommit(自動應答模式自動處理，本例中採用手動應答模式) 
     * @throws Exception 
     */  
    public static void txRecv() throws Exception{  
        Connection conn=MqManager.newConnection();  
        Channel channel=conn.createChannel();  
        //開啟transaction機制  
        channel.txSelect();  
        channel.queueDeclare(queue_name,false,false,true,null);  
        //關閉自動應答模式(自動應答模式不需要ack、txCommit)，需要手動basicAck，並執行txCommit  
        channel.basicConsume(queue_name, false, new DefaultConsumer(channel){  
            @Override  
            public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException{  
                String mes=SerializationUtils.deserialize(body);  
                System.out.println("tx Received :'"+mes+"' done");  
                channel.basicAck(envelope.getDeliveryTag(), false);  
                channel.txCommit();  
                  
            }  
        });  
    }  
      
    public static void main(String[] args) throws Exception {  
        txSend("hello world!");  
        txRecv();  
    }  
}
  • 應答模式之confirm機制
1、確認publisher傳送訊息到broker，由broker進行應答(不能確認是否被有效消費) 
2、confirmSelect，進入confirm訊息確認模式，確認方式：1、非同步ConfirmListener；2、同步waitForConfirms 
3、ConfirmListener、waitForConfirms均需要配合confirm機制使用 
4、暫時未弄明白confirm機制在consumer的應用，ConfirmListener在consumer中無效
5、basicNack、basicReject：引數requeue=true時，訊息會重新進入佇列 
6、autoDelete佇列在消費者關閉後不管是否還有未處理的訊息都會關閉掉
package com.demo.mq.rabbitmq.example11;  
import java.io.IOException;  
import java.io.Serializable;  
import org.apache.commons.lang3.SerializationUtils;  
import com.demo.mq.rabbitmq.MqManager;  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.ConfirmListener;  
import com.rabbitmq.client.Connection;  
  
/** 
 * 應答模式之confirm機制：訊息傳送 
 * @author sheungxin 
 * 
 */  
public class ConfirmSend {  
    private static String exchange_name="";  
    private static String queue_name="tx_queue";  
      
    /** 
     * confirm機制：確認publisher傳送訊息到broker，由broker進行應答(不能確認是否被有效消費) 
     * confirmSelect，進入confirm訊息確認模式，確認方式：1、非同步ConfirmListener；2、同步waitForConfirms 
     * ConfirmListener、waitForConfirms均需要配合confirm機制使用 
     * @param mes 
     * @throws Exception 
     */  
    public static void txSend(Serializable mes) throws Exception{  
        Connection conn=MqManager.newConnection();  
        Channel channel=conn.createChannel();  
        //開啟transaction機制  
        channel.confirmSelect();  
        channel.queueDeclare(queue_name,false,false,true,null);  
        //非同步實現傳送訊息的確認(此部分的訊息確認是指傳送訊息到佇列，並非確認訊息的有效消費)  
        channel.addConfirmListener(new ConfirmListener() {  
              
            @Override  
            public void handleNack(long deliveryTag, boolean multiple)  
                    throws IOException {  
                //multiple：測試發現multiple隨機true或false，原因未知  
                System.out.println("Nack deliveryTag:"+deliveryTag+",multiple:"+multiple);  
            }  
              
            @Override  
            public void handleAck(long deliveryTag, boolean multiple)  
                    throws IOException {  
                System.out.println("Ack deliveryTag:"+deliveryTag+",multiple:"+multiple);  
            }  
        });  
        for(int i=0;i<10;i++){  
            channel.basicPublish(exchange_name, queue_name, null, SerializationUtils.serialize(mes.toString()+i));  
        }  
//      channel.waitForConfirms();//同步實現傳送訊息的確認  
        System.out.println("-----------");  
        channel.close();  
        conn.close();  
    }  
      
    public static void main(String[] args) throws Exception {  
        txSend("hello world!");  
    }  
}  
package com.demo.mq.rabbitmq.example11;  
import java.io.IOException;  
import org.apache.commons.lang3.SerializationUtils;  
import com.demo.mq.rabbitmq.MqManager;  
import com.rabbitmq.client.AMQP;  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.DefaultConsumer;  
import com.rabbitmq.client.Envelope;  
  
/** 
 * 應答模式之confirm機制：訊息接收 
 * @author sheungxin 
 * 
 */  
public class ConfirmRecv {  
    private static String queue_name="tx_queue";  
  
    /** 
     * confirm機制：暫時未弄明白confirm機制在consumer的應用，ConfirmListener在consumer中無效 
     * basicNack、basicReject：引數requeue=true時，訊息會重新進入佇列 
     * autoDelete佇列在消費者關閉後不管是否還有未處理的訊息都會關閉掉 
     * @throws Exception 
     */  
    public static void txRecv() throws Exception{  
        Connection conn=MqManager.newConnection();  
        Channel channel=conn.createChannel();  
        //開啟transaction機制  
//      channel.confirmSelect();  
        //autoDelete,true只要被訊息  
        channel.queueDeclare(queue_name,false,false,true,null);  
        //關閉自動應答模式  
        channel.basicConsume(queue_name, false, new DefaultConsumer(channel){  
            @Override  
            public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException{  
                String mes=SerializationUtils.deserialize(body);  
                //multiple批量提交，true提交小於引數中tag訊息  
                long n=envelope.getDeliveryTag()%3;  
                if(n==0){  
                    channel.basicAck(envelope.getDeliveryTag(), false);  
                }else if(n==1){  
                    //requeue，true重新進入佇列  
                    channel.basicNack(envelope.getDeliveryTag(), false, true);  
                }else{  
                    //requeue，true重新進入佇列,與basicNack差異缺少multiple引數  
                    channel.basicReject(envelope.getDeliveryTag(), true);  
                }  
                try {  
                    Thread.sleep(2*1000);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
                System.out.println((n==0?"Ack":n==1?"Nack":"Reject")+" mes :'"+mes+"' done");  
            }  
        });  
    }  
      
    public static void main(String[] args) throws Exception {  
        txRecv();  
    }  
}

http://blog.csdn.net/azhegps/article/details/53815201

相關推薦

RabbitMQ使用場景練習：訊息確認機制(十一)

訊息確認機制RabbitMQ提供了transaction、confirm兩種訊息確認機制。transaction即事務機制，手動提交和回滾；confirm機制提供了Confirmlistener和waitForConfirms兩種方式。confirm機制效率明顯會高於tran

RabbitMQ訊息佇列（九）：Publisher的訊息確認機制

       在前面的文章中提到了queue和consumer之間的訊息確認機制：通過設定ack。那麼Publisher能不到知道他post的Message有沒有到達queue，甚至更近一步，是否被某個

中介軟體系列十 RabbitMQ之消費者端的訊息確認機制

概述 在RabbitMQ中，即使將queue，exchange, message等都設定了持久化之後，還是不能保證100%保證資料不丟失了。為了實現訊息不丟失，我們需要從Consumer端和Productor端同時進行處理。本篇文章先介紹Consumer端，在AMPQ-0-

【RabbitMq】rabbitMq訊息確認機制

  一、提出問題 生產者將訊息傳送出去後，訊息是否到達RabbitMq伺服器呢？預設的情況下，是不知道的 二、引入訊息確認機制 兩種方式：           1.AMQP實現事務機制     &

java rabbitmq ack訊息確認機制

ackage com.example.demo.ConsumerDemo; import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframewo

（三）rabbitmq的訊息確認機制ack

接上一篇文章，在application配置檔案中新增如下配置： ## 訊息手動確認 spring.rabbitmq.listener.simple.acknowledge-mode=manual 這樣就開啟了訊息手動確認，然後再消費者端程式碼中加上如下程式碼進行業務處理完後的訊息確認刪除訊息

RabbitMQ 之訊息確認機制（事務+Confirm）

概述 在 Rabbitmq 中我們可以通過持久化來解決因為伺服器異常而導致丟失的問題,除此之外我們還會遇到一個問題:生產者將訊息傳送出去之後,訊息到底有沒有正確到達 Rabbit 伺服器呢?如果不錯得數處理,我們是不知道的,(即 Rabbit 伺服器不會反饋任何訊息給生產者),也就是預設的情況下是不知道訊息

RabbitMQ之訊息確認機制（事務+Confirm）

概述 在使用RabbitMQ的時候，我們可以通過訊息持久化操作來解決因為伺服器的異常奔潰導致的訊息丟失，除此之外我們還會遇到一個問題，當訊息的釋出者在將訊息傳送出去之後，訊息到底有沒有正確到達broker代理伺服器呢？如果不進行特殊配置的話，預設情況下發布操作是不會返回任何

RabbitMQ 訊息確認機制 以及 原理解析

一、場景   當訊息的投送方把訊息投遞出去，卻不知道訊息是否投遞成功了。如果訊息投送方不管的話，勢必對系統的造成可靠性的影響。 可是如果要保證系統的可靠性，訊息投靠方，如何知道訊息是否投放成功了呢？ 這個就需要訊息的確認機制，我們來看下rabbitMQ的訊息

RabbitMQ ——與SpringBoot整合並實現訊息確認機制

不囉嗦直接上程式碼 目錄結構如下： pom.xml <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instanc

spring rabbitmq 訊息確認機制和事務支援

確認並且保證訊息被送達，提供了兩種方式：釋出確認和事務。(兩者不可同時使用)在channel為事務時，不可引入確認模式；同樣channel為確認模式下，不可使用事務。 釋出確認： Confirms給客戶端一種輕量級的方式，能夠跟蹤哪些訊息被broker處理，哪些可能因為br

RabbitMQ之訊息確認機制AMQP事務

概述 我們在RabbitMQ中可以通過持久化來解決伺服器掛掉而丟失資料問題,但是大家有沒有想過,我的訊息到達了RabbitMQ伺服器了嗎??? 我們是不知道的,導致的問題就是 如果訊息在到達伺服器之前就丟失了,持久化也是不能解決問題的! 那怎麼辦? 我

RabbitMQ 訊息持久化、事務、Publisher的訊息確認機制

RabbitMQ  訊息持久化、事務、Publisher的訊息確認機制 1. 宣告MessageQueue 在RabbitMQ中，無論是生產者傳送訊息還是消費者接受訊息，都首先需要宣告一個MessageQueue。 這就存在一個問題，是生產者宣告還是消費者宣告呢？要解決這個

RabbitMQ使用場景練習：延遲佇列(八)

延時佇列     在實際業務場景中可能會用到延時訊息傳送，例如支付場景，準時支付、超過未支付將執行不同的方案，其中超時未支付可以看做一個延時訊息。      RabbitMQ本身不具有延時訊息佇列的功能，但是可以通過TTL(Time To Live)、DLX(Dead Le

RabbitMQ ACK 訊息確認機制

注意如果拋異常或unack（並且requeue為true），訊息會一直重新入佇列，一不小心就會xxxxx一大堆訊息不斷重複~。 //訊息的標識，false只確認當前一個訊息收到，true確認所有consumer獲得的訊息 (正常消費) channel.basicAck(message.getMess

RabbitMQ使用場景練習：監聽器Listener(九)

監聽器     RabbitMQ中監聽器有ReturnListener、ConfirmListener、ShutdownListener，本練習中使用ReturnListener，在釋出訊息時設定mandatory等於true，監聽訊息是否有相匹配的佇列，沒有時Return

RabbitMQ學習(六)——訊息確認機制(Confirm模式)

在上一篇文章中我們講解了RabbitMQ中的AMQP事務來保證訊息傳送到Broker端，同時我們可以在事務之間傳送多條訊息（即在channel.txSelect()和channel.txCommit()之間傳送多條訊息，通過使用事務來保證它們準確到達Broker

RabbitMQ學習(五)——訊息確認機制(AMQP事務)

在前面的文章中，我們對RabbitMQ的訊息分發機制做了探究，知道RabbitMQ訊息的分發機制，包括公平分發和輪詢分發，如果忘記了可以去複寫一下RabbitMQ學習(四)——訊息分發機制。 我們知道可以通過持久化（交換機、佇列和訊息持久化）來保障我們在伺服器崩

RabbitMQ實戰-訊息確認機制之訊息的正確消費

上節中我們講了如何確保訊息的準確釋出，今天我們來看看如何確保訊息的正確消費。 在之前的基礎上我們對消費者（倉庫服務）進行完善。 修改配置檔案application.yml 消費者的ack方式預設是自動的，也就是說訊息一旦被消費（無論是否處理成功），訊息都會被確認，然後會從

九、rabbitMQ的訊息確認機制之事務機制

說明：在rabbitMQ中，我們為了解決伺服器異常導致資料丟失的問題，我們可以採用rabbitMQ的持久化機制，但是我們如何確定生產者將訊息傳送給了rabbitMQ呢，那麼我們採用兩種協議的模式。    （1）、AMQP實現了事務機制    （2）、confirm模式一、事務