【README】 業(yè)務(wù)場(chǎng)景：? 業(yè)務(wù)處理伴隨消息的發(fā)送，業(yè)務(wù)處理失敗（事務(wù)回滾）后要求消息不發(fā)送。

補(bǔ)充1：ACK與CONFIRM的區(qū)別

ACK-消費(fèi)者消費(fèi)成功后確認(rèn)；（消費(fèi)者確認(rèn)已收到）

CONFIRM-事務(wù)生產(chǎn)者異常事務(wù)回滾，不發(fā)送到mq服務(wù)器代理;（生產(chǎn)者確認(rèn)發(fā)送成功）

補(bǔ)充2： 參考了這篇文章， 寫(xiě)的非常棒； https://blog.csdn.net/u013256816/article/details/55515234/

?

【1】交換機(jī)，隊(duì)列聲明與綁定

/*** 交換機(jī)，隊(duì)列聲明與綁定 */ public class ConfirmDeclarer {/** 確認(rèn)交換機(jī) */public static final String CONFIRM_EXCHANGE2 = "CONFIRM_EXCHNAGE2";/** 確認(rèn)隊(duì)列 */public static final String CONFIRM_QUEUE2 = "CONFIRM_QUEUE2";/** 路由鍵 */public static final String CONFIRM_ROUTE2 = "CONFIRM_ROUTE2";// 四種Exchange 模式 // direct ：需要生產(chǎn)者和消費(fèi)者綁定相同的Exchange和routing key。 // fanout：廣播模式需要生產(chǎn)者消費(fèi)者綁定相同的Exchange。 // topic：支持模糊匹配的廣播模式以點(diǎn)分隔，*表示一個(gè)單詞，#表示任意數(shù)量（零個(gè)或多個(gè)）單詞。 // header：根據(jù)生產(chǎn)者和消費(fèi)者的header中信息進(jìn)行匹配性能較差 ，x-match [all 匹配所有/any 任意一個(gè)]。public static void main(String[] args) throws Exception {/* 獲取連接*/Connection conn = RBConnectionUtil.getConn();// 創(chuàng)建信道 Channel channel = conn.createChannel(); /* 聲明交換機(jī) */channel.exchangeDeclare(CONFIRM_EXCHANGE2, BuiltinExchangeType.FANOUT);/* 聲明隊(duì)列 */channel.queueDeclare(CONFIRM_QUEUE2, true, false, false, null); System.out.println(String.format("聲明交換機(jī)【%s】，隊(duì)列【%s】成功", CONFIRM_EXCHANGE2, CONFIRM_QUEUE2));/* 把隊(duì)列綁定到交換機(jī) */channel.queueBind(CONFIRM_QUEUE2, CONFIRM_EXCHANGE2, CONFIRM_ROUTE2);/* 關(guān)閉信道和連接 */channel.close();conn.close(); } }

? 【2】生產(chǎn)者，基于事務(wù)實(shí)現(xiàn)消息確認(rèn)

/*** 消息確認(rèn)生產(chǎn)者*/ public class ConfirmProducer {public static void main(String[] args) throws Exception {/* 獲取連接*/Connection conn = RBConnectionUtil.getConn();// 創(chuàng)建信道 Channel channel = conn.createChannel();String[] messages = new String[]{"first.-04130828", "second..-04130828", "third...-04130828", "fourth....-04130828", "fiveth.....-04130828", "6th.....-04130828", "7th.....-04130828", "8th.....-04130828", "9th.....-04130828", "10th.....-04130828"};channel.txSelect(); // 開(kāi)啟事務(wù)int i=0;for (String msg : messages) {try {send(channel, msg); // 發(fā)送消息 System.out.println(msg + " is sent");if (++i > 5) {int temp = 1/0; // 拋出0除異常 } channel.txCommit(); // 提交事務(wù) System.out.println("消息發(fā)送完成，事務(wù)提交，消息=" + msg); } catch(Exception e) { channel.txRollback(); // 拋出異常， 回滾事務(wù)System.out.println("拋出異常，消息發(fā)送取消，事務(wù)回滾，消息=" + msg); }}channel.close(); conn.close();}private static void send(Channel channel, String msg) throws Exception { // 發(fā)送消息channel.basicPublish(ConfirmDeclarer.CONFIRM_EXCHANGE2, ConfirmDeclarer.CONFIRM_ROUTE2, null, msg.getBytes());} }

【3】日志

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". SLF4J: Defaulting to no-operation (NOP) logger implementation SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. first.-04130828 is sent 消息發(fā)送完成，事務(wù)提交，消息=first.-04130828 second..-04130828 is sent 消息發(fā)送完成，事務(wù)提交，消息=second..-04130828 third...-04130828 is sent 消息發(fā)送完成，事務(wù)提交，消息=third...-04130828 fourth....-04130828 is sent 消息發(fā)送完成，事務(wù)提交，消息=fourth....-04130828 fiveth.....-04130828 is sent 消息發(fā)送完成，事務(wù)提交，消息=fiveth.....-04130828 6th.....-04130828 is sent 拋出異常，消息發(fā)送取消，事務(wù)回滾，消息=6th.....-04130828 7th.....-04130828 is sent 拋出異常，消息發(fā)送取消，事務(wù)回滾，消息=7th.....-04130828 8th.....-04130828 is sent 拋出異常，消息發(fā)送取消，事務(wù)回滾，消息=8th.....-04130828 9th.....-04130828 is sent 拋出異常，消息發(fā)送取消，事務(wù)回滾，消息=9th.....-04130828 10th.....-04130828 is sent 拋出異常，消息發(fā)送取消，事務(wù)回滾，消息=10th.....-04130828

【4】rabbitmq首頁(yè)查看隊(duì)列中的消息總數(shù)

?

總結(jié)

以上是生活随笔為你收集整理的rabbitmq生产者基于事务实现发送确认的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。