2. 程式人生 > >RabbitMq + Spring 實現ACK機制

RabbitMq + Spring 實現ACK機制

阿新 • • 發佈：2018-12-14

概念性解讀(Ack的靈活)

         首先啊，有的人不是太理解這個Ack是什麼，講的接地氣一點，其實就是一個通知，怎麼說呢，當我監聽消費者，正常情況下，不會出異常，但是如果是出現了異常，甚至是沒有獲取的異常，那是不是這條資料就會作廢，但是我們肯定不希望這樣的情況出現，我們想要的是，如果在出現異常的時候，我們識別到，如果確實是一個不良異常，肯定希望資料重新返回佇列中，再次執行我們的業務邏輯程式碼，此時我就需要一個Ack的通知，告訴佇列服務，我是否已經成功處理了這條資料，而如果不配置Ack的話呢，我測試過他會自動的忽略，也就是說此時的服務是no_ack=true的模式，就是說只要我發現你是消費了這個資料，至於異常不異常的，我不管了。通知Ack機制就是這麼來的，更加靈活的，我們需要Ack不自動，而是手動，這樣做的好處，就是使得我們開發人員更加人性化或者靈活的來處理我們的業務羅傑程式碼，更加方便的處理異常的問題以及資料的返回處理等。下面是通話機制的四條原則：

  • Basic.Ack 發回給 RabbitMQ 以告知，可以將相應 message 從 RabbitMQ 的訊息快取中移除。
  • Basic.Ack 未被 consumer 發回給 RabbitMQ 前出現了異常，RabbitMQ 發現與該 consumer 對應的連線被斷開，之後將該 message 以輪詢方式傳送給其他 consumer （假設存在多個 consumer 訂閱同一個 queue）。
  • 在 no_ack=true 的情況下，RabbitMQ 認為 message 一旦被 deliver 出去了，就已被確認了，所以會立即將快取中的 message 刪除。所以在 consumer 異常時會導致訊息丟失。
  • 來自 consumer 側的 Basic.Ack 與 傳送給 Producer 側的 Basic.Ack 沒有直接關係。

正題部分(配置手動Ack，實現異常訊息回滾)

A. 在消費者端的mq配置檔案上新增,配置  關鍵程式碼為 acknowledeg = "manual"，意為表示該消費者的ack方式為手動（此時的queue已經和生產者的exchange通過某個routeKey綁定了）

<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
    <rabbit:listener queues="queue_xxx" ref="MqConsumer"/>
    <rabbit:listener queues="queue_xxx" ref="MqConsumer2"/>
</rabbit:listener-container>

B. 新建一個類 MqConsumer ，並實現介面  ChannelAwareMessageListener ，實現onMessage方法，不需要指定方法。

springAMQP中已經實現了一個功能，如果該監聽器已經實現了下面2個介面，則直接呼叫onMessage方法

C. 關鍵點在實現了ChannelAwareMessageListener的onMessage方法後，會有2個引數。

一個是message（訊息實體），一個是channel就是當前的通道，很多地方都沒有說清楚怎麼去手動ack，其實手動ack就是在當前channel裡面呼叫basicAsk的方法，並傳入當前訊息的tagId就可以了。

channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

其中deliveryTag是tag的id，由生產者生成。第二個引數我其實也沒理解用途，暫時還沒有模擬出場景，所以先不討論。

同樣的，如果要Nack或者拒絕訊息（reject）的時候，也是呼叫channel裡面的basicXXX方法就可以了（當然要制定tagId）。注意如果拋異常或Nack（並且requeue為true），訊息會一直重新入佇列，一不小心就會重複一大堆訊息不斷出現~。

channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); // 訊息的標識，false只確認當前一個訊息收到，true確認所有consumer獲得的訊息
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); // ack返回false，並重新回到佇列，api裡面解釋得很清楚
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true); // 拒絕訊息

D. 針對上面所描述的情況，我們在搭建一個訊息佇列時候，我們的思路應該是這樣的，首先，我們要啟動ack的手動方式，緊接著，我們處理程式碼邏輯，如果發生了異常資訊，我們首先通知到ack，我已經表示接受到這條資料了，你可以進行刪除了，不需要讓他自動的重新進入佇列中，然後，我們啟用一個錯誤處理，手動將其重新插入佇列中，在此之前，有幾個類和Api一起來看一下。

    1. SimpleMessageListenerContainer

    這個是我們的基礎監聽，他的作用就是佇列的總監聽，可以為其配置ack模式，異常處理類等。。

    2. org.springframework.amqp.support.converter.SimpleMessageConverter

    這個類和下面的Converter類很容易搞混淆，這個類的作用是可以解析佇列中的 message 轉 obj

    3. org.springframework.amqp.rabbit.retry.MessageRecoverer

    這個介面，需要我們開發者自定義實現，其中的一個方法recover(Message message, Throwable cause)，就可以看出來他是幹嘛的，就是說在監聽出錯，也就是沒有抓取的異常而是丟擲的異常會觸發該方法，我們就會在這個介面的實現中，將訊息重新入佇列

    4. org.springframework.util.ErrorHandler

    這個介面也是在出現異常時候，會觸發他的方法

E.  完整例項

    1. spring配置佇列xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:aop="http://www.springframework.org/schema/aop" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
    http://www.springframework.org/schema/rabbit
    http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
    http://www.springframework.org/schema/aop
    http://www.springframework.org/schema/aop/spring-aop-3.1.xsd
    http://www.springframework.org/schema/context
    http://www.springframework.org/schema/context/spring-context-3.1.xsd">

    <!-- 連線服務配置 -->
    <rabbit:connection-factory id="connectionFactory"
        host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}"
        password="${rabbitmq.password}" channel-cache-size="${rabbitmq.channel.cache.size}" />
    
    <!-- 設定Ack模式為手動 -->    
    <bean id="ackManual"
        class="org.springframework.beans.factory.config.FieldRetrievingFactoryBean">
        <property name="staticField"
            value="org.springframework.amqp.core.AcknowledgeMode.MANUAL" />
    </bean>

    <!-- 異常處理，記錄異常資訊 --> 
    <bean id="mqErrorHandler" class="com.zefun.wechat.utils.MQErrorHandler"/>
    <!-- 將類自動注入，可解析msg資訊 -->    
    <bean id="msgConverter" class="org.springframework.amqp.support.converter.SimpleMessageConverter" />

	<!-- 建立rabbitAdmin 代理類 -->
    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>
    <rabbit:admin connection-factory="connectionFactory" />
    
    <!-- 建立SimpleMessageListenerContainer的理想通道，主要實現異常事件處理邏輯 -->
    <bean id="retryOperationsInterceptorFactoryBean"
        class="org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean">
        <property name="messageRecoverer">
            <bean class="com.zefun.wechat.utils.MQRepublishMessageRecoverer"/>
        </property>
        <property name="retryOperations">
            <bean class="org.springframework.retry.support.RetryTemplate">
                <property name="backOffPolicy">
                    <bean
                        class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
                        <property name="initialInterval" value="500" />
                        <property name="multiplier" value="10.0" />
                        <property name="maxInterval" value="10000" />
                    </bean>
                </property>
            </bean>
        </property>
    </bean>
    
    <!-- 定義佇列，在下面的交換機中引用次佇列，實現繫結 -->     
    <rabbit:queue id="queue_system_error_logger_jmail" name="${rabbitmq.system.out.log.error.mail}" durable="true"
        auto-delete="false" exclusive="false" />
                         
    <!--路由設定 將佇列繫結，屬於direct型別 -->
    <rabbit:direct-exchange id="directExchange"
        name="directExchange" durable="true" auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding queue="queue_system_error_logger_jmail" key="${rabbitmq.system.out.log.error.mail}" />
        </rabbit:bindings>
    </rabbit:direct-exchange>
    
    
    <!-- logger 日誌傳送功能 -->
    <bean class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="acknowledgeMode" ref="ackManual" />
        <property name="queueNames" value="${rabbitmq.system.out.log.error.mail}" />
        <property name="messageListener">
            <bean class="com.zefun.wechat.listener.SystemOutLogErrorMessageNoitce" />
        </property>
        <property name="concurrentConsumers" value="${rabbitmq.concurrentConsumers}" />
        <property name="adviceChain" ref="retryOperationsInterceptorFactoryBean" />
        <property name="errorHandler" ref="mqErrorHandler" />
    </bean>    
</beans>

    2. MessageRecoverer 配置，將小心重新入佇列

package com.zefun.wechat.utils;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Map;

import org.apache.log4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;

public class MQRepublishMessageRecoverer implements MessageRecoverer {
	
	private static final Logger logger = Logger.getLogger(MQRepublishMessageRecoverer.class);

	@Autowired
	private RabbitTemplate rabbitTemplate;
	
	@Autowired
	private MessageConverter msgConverter;

	@Override
	public void recover(Message message, Throwable cause) {
		Map<String, Object> headers = message.getMessageProperties().getHeaders();
		headers.put("x-exception-stacktrace", getStackTraceAsString(cause));
		headers.put("x-exception-message", cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage());
		headers.put("x-original-exchange", message.getMessageProperties().getReceivedExchange());
		headers.put("x-original-routingKey", message.getMessageProperties().getReceivedRoutingKey());
		this.rabbitTemplate.send(message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), message);
		logger.error("handler msg (" + msgConverter.fromMessage(message) + ") err, republish to mq.", cause);
	}

	private String getStackTraceAsString(Throwable cause) {
		StringWriter stringWriter = new StringWriter();
		PrintWriter printWriter = new PrintWriter(stringWriter, true);
		cause.printStackTrace(printWriter);
		return stringWriter.getBuffer().toString();
	}
}

    3. MQErrorHandler 寫法，在出現異常時，記錄異常

package com.zefun.wechat.utils;

import java.lang.reflect.Field;
import java.util.Date;

import org.apache.commons.lang.reflect.FieldUtils;
import org.apache.log4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.ErrorHandler;

import com.zefun.wechat.service.RedisService;

public class MQErrorHandler implements ErrorHandler {

    private static final Logger logger = Logger.getLogger(MQErrorHandler.class);

    @Autowired
    private RedisService redisService;
    @Autowired
    private MessageConverter msgConverter;

    @Override
    public void handleError(Throwable cause) {
        Field mqMsgField = FieldUtils.getField(MQListenerExecutionFailedException.class, "mqMsg", true);
        if (mqMsgField != null) {
            try {
                Message mqMsg = (Message) mqMsgField.get(cause);
                Object msgObj = msgConverter.fromMessage(mqMsg);
                logger.error("handle MQ msg: " + msgObj + " failed, record it to redis.", cause);
                redisService.zadd(App.MsgErr.MQ_MSG_ERR_RECORD_KEY, new Double(new Date().getTime()), msgObj.toString());
            } catch (Exception e) {
                e.printStackTrace();
            }
        } else {
            logger.error("An error occurred.", cause);
        }
    }

}

    4. SystemOutLogErrorMessageNoitce 實現 ChannelAwareMessageListener介面，處理郵件服務

package com.zefun.wechat.listener;

import javax.mail.internet.MimeMessage;

import org.apache.log4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.mail.javamail.JavaMailSenderImpl;
import org.springframework.mail.javamail.MimeMessageHelper;

import com.rabbitmq.client.Channel;
import com.zefun.wechat.utils.App;
import net.sf.json.JSONObject;

public class SystemOutLogErrorMessageNoitce  implements ChannelAwareMessageListener {

    private static final Logger logger = Logger.getLogger(MemberWechatMessageTextNoitce.class);
    @Autowired
    private MessageConverter msgConverter;
    /** logger b */
    @Autowired
    private JavaMailSenderImpl senderImpl;
    
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {

        Object obj = null;
        try {
            obj = msgConverter.fromMessage(message);
        } catch (MessageConversionException e) {
            logger.error("convert MQ message error.", e);
        } finally {
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            if (deliveryTag != App.DELIVERIED_TAG) {
                channel.basicAck(deliveryTag, false);
                message.getMessageProperties().setDeliveryTag(App.DELIVERIED_TAG);
                logger.info("revice and ack msg: " + (obj == null ? message : new String((byte[]) obj)));
            }
        }
        if (obj == null) {
            return;
        }
        JSONObject map = JSONObject.fromObject(obj);
        sendMailSystemLoggerError(map.getString("date"), map.getString("subject"), map.getString("domain"), map.getString("requestURL"), map.getString("message"));
    }
    
    /**
     * jmail logger 
    * @author 小高
    * @date 2016年10月25日 下午3:24:46
    * @param date          日期
    * @param subject       主題賬戶
    * @param domain        域名環境
    * @param message       logger日誌
    * @param requestURL    請求路徑
    * @throws Exception    異常資訊
     */
    public void sendMailSystemLoggerError(String date, String subject, String domain, String requestURL, String message) throws Exception{
        MimeMessage mailMessage = this.senderImpl.createMimeMessage();
        MimeMessageHelper messageHelper = new MimeMessageHelper(mailMessage, true);
        messageHelper.setTo("[email protected]");
        messageHelper.setFrom("[email protected]");
        messageHelper.setSubject(date + " 系統異常");
        String msg = "<p>異常時間：" + date + "</p><p>門店企業：" + subject + "</p>"
                    + "<p>部署環境：" + domain + "</p><p>異常連線：" + requestURL + "</p>"
                    + "<p>異常內容：</p>" + message;
        messageHelper.setText("<html><head></head><body>" + msg + "</body></html>", true);
        senderImpl.send(mailMessage);
        logger.info("jmail push message success");
    }

}

相關推薦

RabbitMq + Spring 實現ACK機制

概念性解讀(Ack的靈活)          首先啊，有的人不是太理解這個Ack是什麼，講的接地氣一點，其實就是一個通知，怎麼說呢，當我監聽消費者，正常情況下，不會出異常，但是如果是出現了異常，甚至是沒有獲取的異常，那是不是這條資料就會作廢，但是我們肯定不希望這樣的情況

rabbitmq結合springboot實現ACK機制的消費機制

###1.RabbitMQ介紹 RabbitMQ是實現AMQP（高階訊息佇列協議）的訊息中介軟體的一種，最初起源於金融系統，用於在分散式系統中儲存轉發訊息，在易用性、擴充套件性、高可用性等方面表現不俗。RabbitMQ主要是為了實現系統之間的雙向解耦而實現的。

RabbitMQ訊息應答------ack機制

Message acknowledgment（訊息應答） 執行一個任務可能需要花費幾秒鐘，你可能會擔心如果一個消費者在執行任務過程中掛掉了。一旦RabbitMQ將訊息分發給了消費者，就會從記憶體

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

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

RabbitMQ系列之六 Spring RabbitMQ整合實現案例之 非同步郵件傳送

摘要：給使用者傳送郵件的場景，其實也是比較常見的，比如使用者註冊需要郵箱驗證，使用者異地登入傳送郵件通知等等，在這裡我以 RabbitMQ 實現非同步傳送郵件。 專案git地址：https://github.com/gitcaiqing/RabbitMQ-Email 1.專案結構 2.

Spring Cloud Stream RabbitMQ 簡單實現

參考官方文件:http://cloud.spring.io/spring-cloud-static/Finchley.SR1/single/spring-cloud.html#_rabbitmq_binder 環境：win10 rabbit 預設配置 一、新增依賴 <depende

RabbitMQ ——Spring整合及exchange的direct、topic方式實現和簡單佇列實現

程式整體結構 Maven依賴 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schema

6.Spring RabbitMQ整合實現案例之 非同步郵件傳送

摘要：給使用者傳送郵件的場景，其實也是比較常見的，比如使用者註冊需要郵箱驗證，使用者異地登入傳送郵件通知等等，在這裡我以 RabbitMQ 實現非同步傳送郵件。 專案git地址：https://github.com/gitcaiqing/RabbitMQ-Ema

Spring事務管理機制的實現原理-動態代理

之前在做專案中遇到spring無法進行事務代理問題，最後發現是因為沒有寫介面，原因當時明白了，看到這篇文章寫的清楚些，轉過來  我們先來分析一下Spring事務管理機制的實現原理。由於Spring內建AOP預設使用動態代理模式實現，我們就先來分析一下動態代理模式的實現方 法。動態代理模式的核心就在於程

35. Spring Boot整合Redis實現快取機制【從零開始學Spring Boot】

【視訊&交流平臺】 http://study.163.com/course/introduction.htm?courseId=1004329008&utm_campaign=commission&utm_source=40000000

ack是什麼，如何使用Ack機制，如何關閉Ack機制，基本實現，STORM的訊息容錯機制，Ack機制

1、ack是什麼 ack 機制是storm整個技術體系中非常閃亮的一個創新點。 通過Ack機制，spout傳送出去的每一條訊息，都可以確定是被成功處理或失敗處理， 從而可以讓開發者採取動作。比如在Meta中，成功被處理，即可更新偏移量，當失敗時，重複傳送資料

RabbitMQ的架構、交換機機制和ACK機制

RabbitMQ  RabbitMQ是一個由erlang開發的AMQP（Advanced Message Queue ）的開源實現。 RabbitMQ 即一個訊息佇列，_主要是用來實現應用程式的非同步和解耦，同時也能起到訊息緩衝，訊息分發的作用。_ RabbitMQ使用

spring-boot整合ehcache實現快取機制

EhCache 是一個純Java的程序內快取框架，具有快速、精幹等特點，是Hibernate中預設的CacheProvider。 ehcache提供了多種快取策略，主要分為記憶體和磁碟兩級，所以無需擔心容量問題。 spring-boot是一個快速的整合框架，其設計目的

Spring Boot整合Redis實現快取機制

本文章牽涉到的技術點比較多： 、Redis、Spring MVC,Spirng Cache，所以在看這篇文章的時候，需要對以上這些技術點有一定的瞭解或者也可以先看看這篇文章，針對文章中實際的技術點在進一步瞭解（注意，您需要自己下載Redis Server到您的本地，

如何使用Spring Boot與RabbitMQ結合實現延遲佇列

背景 何為延遲佇列？ 顧名思義，延遲佇列就是進入該佇列的訊息會被延遲消費的佇列。而一般的佇列，訊息一旦入隊了之後就會被消費者馬上消費。 延遲佇列能做什麼？延遲佇列多用於需要延遲工作的場景。最常見的是以下兩種場景：延遲消費。比如： 使用者生成訂單之後

深入剖析 RabbitMQ —— Spring 框架下實現 AMQP 高階訊息佇列協議

前言 訊息佇列在現今資料量超大，併發量超高的系統中是十分常用的。本文將會對現時最常用到的幾款訊息佇列框架 ActiveMQ、RabbitMQ、Kafka 進行分析對比。詳細介紹 RabbitMQ 在 Sprinig 框架下的結構及實現原理，從Producer 端的事務、回撥函式（ConfirmCa

Spring事務的機制

pri 異常捕獲 for cep 顯式 nbsp class transacti .class 默認spring事務只在發生未被捕獲的 runtimeexcetpion時才回滾。 spring aop 異常捕獲原理：被攔截的方法需顯式拋出異常，並不能經任何處理，這樣ao

使用spring實現讀寫分離

item ger [] 依據 sign batis repl arch 定義 1. 背景 我們一般應用對數據庫而言都是“讀多寫少”，也就說對數據庫讀取數據的壓力比較大，有一個思路就是說采用數據庫集群的方案， 其中一個是主庫，負責寫入數據，我們稱之為：寫庫； 其它都是從庫，

PHP實現事件機制實例分析

word-wrap [0 popu except switch targe att otto 對象 PHP實現事件機制實例分析 內置了事件機制的語言不多，php也沒有提供這種功能。事件(Event)說簡單了就是一個Observer模式。實現起來非常ea

Spring 實現部分事務回滾

light back true prop 回滾 sage .class lba aaa 例如有業務需求，在catch異常後，catch塊內把異常的信息存入到數據庫，而catch外的數據全部回滾 try { ....... aaaService.save();