手動應答模式(manual)

解釋:
????????手動應答:既是當消費者消費了隊列中消息時需要給隊列一個應答,告訴隊列這條消息我已經(jīng)消費了,可以刪除了;
? ? ? ? 若是不應答,即使消費了 隊列沒收到消費成功的提示 所有消息會一直在隊列中;
? ??注意 注意 注意:重要的事情說三遍,下面說的很重要
?場景：

????????當我們使用了手動應答模式,消費者若是成功消費了信息,我們給隊列一個成功應答(channel.basicAck(deliveryTag,false);),然后隊列收到應答后就會把此消息刪除,這點時毋庸置疑的,因為我們已成功消費了這個消息,也不想讓此消息繼續(xù)留在隊列中;
????????但是,若是消費者消費消息失敗了,該怎么辦? 這時不能再給隊列一個成功的應答(給了那這條消息不就丟失了嗎),不給應答呢 消息又一直在隊列中，而倘若我們給一個拒絕應答(channel.basicReject(deliveryTag,true) true會重回隊列,若是false 消息就丟了,一般都會設置為true;)，那么你就成功掉進坑里了,這樣隊列會一直循環(huán)投遞消息,而消費者這邊又不能成功消費,消費者又拒絕應答,隊列又投遞消息......
? ? ? ? 對,此時就會進入死循環(huán),搞不好光錯誤日志就會沾滿內存;這時該有人會說了,不對,我們在配置文件中不是開啟了重試,并且配置了最大重試次數(shù)了嗎?
? ? ? ? 如下配置:
? ? ?* ? ? ? ? ? ? ? ? ?#是否開啟自動重試 默認為false 不開啟
? ? ?* ? ? ? ? ? ? ? ? ?spring.rabbitmq.listener.simple.retry.enabled=true
? ? ?* ? ? ? ? ? ? ? ? ?#最大重試次數(shù)
? ? ?* ? ? ? ? ? ? ? ? ?spring.rabbitmq.listener.simple.retry.max-attempts=5
? ? ?* ? ? ? ? ? ? ? ? ?#最大重試時間間隔
? ? ?* ? ? ? ? ? ? ? ? ?spring.rabbitmq.listener.simple.retry.max-interval=20000ms
? ? ?* ? ? ? ? ? ? ? ? ?#重試時間間隔
? ? ?* ? ? ? ? ? ? ? ? ?spring.rabbitmq.listener.simple.retry.initial-interval=2000ms
? ? ?* ? ? ? ? ? ? ? ? ?# 最大重試間隔*乘數(shù)
? ? ?* ? ? ? ? ? ? ? ? ?#應用于上一重試間隔的乘數(shù) 第一次(重試時間間隔)2s 4s 8s 16s 32s 此處32s>20s 以后都以20s為間隔 總的次數(shù)為最大重試次數(shù)
? ? ?* ? ? ? ? ? ? ? ? ?spring.rabbitmq.listener.simple.retry.multiplier=2
????????那么,應該會在最大重試次數(shù)試完還不成功就應該不嘗試投遞了啊,這樣不應該走死循環(huán)了啊!
????????其實這就是要特別注意的點,當我們開啟了手動應答時 當消息沒被成功消費并給隊列拒絕時上面那些配置其實都已經(jīng)失效了!!! 所以就會進入死循環(huán)！
????????但是在開發(fā)中我們不想讓消息丟失,那么開啟自動應答顯然不合理,而開啟手動應答時當出現(xiàn)錯誤時(沒成功消費)又會進入死循環(huán),那該怎么解決呢?

解決:

? ? ? ? 引入死信隊列; 當消息沒被成功消費時 我們把這條消息投遞到私信隊列中,然后我們再人為的干預處理此消息即可！

自動應答:

????????顧名思義 當消費者成功消費了隊列中消息,隊列就會自動的把此消息從隊列中刪除,若是沒有消費者或者消費者消費失敗,隊列在嘗試最大重試次數(shù)后就會把此消息刪除;

?注意：

手動模式 若消息沒成功消費時,若給隊列拒絕(重試機制不生效),則會進入死循環(huán),若不給拒絕 拋出異常(重試機制生效) 則消息會堆積在隊列中,后續(xù)投遞的消息也會堆積并不會被消費

/*** 消費者*/@Component
@Slf4j
public class DirectConsumer {/*** 手動應答模式(manual)* 解釋:*   手動應答:既是當消費者消費了隊列中消息時需要給隊列一個應答,告訴隊列這條消息我已經(jīng)消費了,可以刪除了;*       若是不應答,即使消費了 隊列沒收到消費成功的提示 所有消息會一直在隊列中;*    注意 注意 注意:重要的事情說三遍,下面說的很重要*        場景：當我們使用了手動應答模式,消費者若是成功消費了信息,我們給隊列一個成功應答(channel.basicAck(deliveryTag,false);),*            然后隊列收到應答后就會把此消息刪除,這點時毋庸置疑的,因為我們已成功消費了這個消息,也不想讓此消息繼續(xù)留在隊列中;*            但是,若是消費者消費消息失敗了,該怎么辦? 這時不能再給隊列一個成功的應答(給了那這條消息不就丟失了嗎),不給應答呢 消息又一直在隊列中*            而倘若我們給一個拒絕應答(channel.basicReject(deliveryTag,true) true會重回隊列,若是false 消息就丟了,一般都會設置為true;)*            那么你就成功掉進坑里了,這樣隊列會一直循環(huán)投遞消息,而消費者這邊又不能成功消費,消費者又拒絕應答,隊列又投遞消息......*            對,此時就會進入死循環(huán),搞不好光錯誤日志就會沾滿內存;這時該有人會說了,不對,我們在配置文件中不是開啟了重試,并且配置了最大重試次數(shù)了嗎?*            如下配置:*                  #是否開啟自動重試 默認為false 不開啟*                  spring.rabbitmq.listener.simple.retry.enabled=true*                  #最大重試次數(shù)*                  spring.rabbitmq.listener.simple.retry.max-attempts=5*                  #最大重試時間間隔*                  spring.rabbitmq.listener.simple.retry.max-interval=20000ms*                  #重試時間間隔*                  spring.rabbitmq.listener.simple.retry.initial-interval=2000ms*                  # 最大重試間隔*乘數(shù)*                  #應用于上一重試間隔的乘數(shù) 第一次(重試時間間隔)2s 4s 8s 16s 32s 此處32s>20s 以后都以20s為間隔 總的次數(shù)為最大重試次數(shù)*                  spring.rabbitmq.listener.simple.retry.multiplier=2*             那么,應該會在最大重試次數(shù)試完還不成功就應該不嘗試投遞了啊,這樣不應該走死循環(huán)了啊!*             其實這就是要特別注意的點,當我們開啟了手動應答時 當消息沒被成功消費并給隊列拒絕時上面那些配置其實都已經(jīng)失效了!!! 所以就會進入死循環(huán)！*             但是在開發(fā)中我們不想讓消息丟失,那么開啟自動應答顯然不合理,而開啟手動應答時當出現(xiàn)錯誤時(沒成功消費)又會進入死循環(huán),那該怎么解決呢?**             解決: 引入死信隊列; 當消息沒被成功消費時 我們把這條消息投遞到私信隊列中,然后我們再人為的干預處理此消息即可！***   自動應答:顧名思義 當消費者成功消費了隊列中消息,隊列就會自動的把此消息從隊列中刪除,若是沒有消費者或者消費者*        消費失敗,隊列在嘗試最大重試次數(shù)后就會把此消息刪除;** 注解含義：*   1、@RabbitHandler  handler真正的執(zhí)行者*   2、@RabbitListener 監(jiān)聽DirectQueue-01這個隊列** @param user 接受的消息類型為user(生產(chǎn)者發(fā)送的為user類型)* @param message* @param channel* @throws IOException*/@RabbitHandler@RabbitListener(queues = RabbitConfig.QUEUE_KEY_03)public void process2(User user, Message message, Channel channel) throws IOException {
//        long deliveryTag = message.getMessageProperties().getDeliveryTag();try {//業(yè)務開始if (user.getId().equals(5)) {int a=1/0;}System.out.println("接受到消息,并正常處理結束"+ JSONUtil.toJsonStr(user));//業(yè)務結束/*** 確認應答* basicAck(long deliveryTag, boolean multiple)* deliveryTag:當前消息在隊列中的的索引;* multiple:為true的話就是批量確認 是消費一個就應答還是一批處理完再應答；通常都是false 一個一個應答*///           channel.basicAck(deliveryTag,false);}catch (Exception ex){System.out.println(ex.getMessage());System.out.println("接受到消息,發(fā)生異常"+ JSONUtil.toJsonStr(user));System.out.println(user);throw ex;//拒絕    true的時候拒絕,false時消息就丟了
//           channel.basicReject(deliveryTag,true);}}
}