网站建设能否建立国外站,泰州哪家网做网站,动漫制作专业介绍,免费主题网站【README】 业务场景#xff1a; 业务处理伴随消息的发送#xff0c;业务处理失败#xff08;事务回滚#xff09;后要求消息不发送。
补充1#xff1a;ACK与CONFIRM的区别
ACK-消费者消费成功后确认#xff1b;#xff08;消费者确认已收到#xff09;
CONFIRM-事…【README】 业务场景 业务处理伴随消息的发送业务处理失败事务回滚后要求消息不发送。
补充1ACK与CONFIRM的区别
ACK-消费者消费成功后确认消费者确认已收到
CONFIRM-事务生产者异常事务回滚不发送到mq服务器代理;生产者确认发送成功
补充2 参考了这篇文章 写的非常棒 https://blog.csdn.net/u013256816/article/details/55515234/ 【1】交换机队列声明与绑定
/*** 交换机队列声明与绑定 */
public class ConfirmDeclarer {/** 确认交换机 */public static final String CONFIRM_EXCHANGE2 CONFIRM_EXCHNAGE2;/** 确认队列 */public static final String CONFIRM_QUEUE2 CONFIRM_QUEUE2;/** 路由键 */public static final String CONFIRM_ROUTE2 CONFIRM_ROUTE2;// 四种Exchange 模式
// direct 需要生产者和消费者绑定相同的Exchange和routing key。
// fanout广播模式需要生产者消费者绑定相同的Exchange。
// topic支持模糊匹配的广播模式以点分隔*表示一个单词#表示任意数量零个或多个单词。
// header根据生产者和消费者的header中信息进行匹配性能较差 x-match [all 匹配所有/any 任意一个]。public static void main(String[] args) throws Exception {/* 获取连接*/Connection conn RBConnectionUtil.getConn();// 创建信道 Channel channel conn.createChannel(); /* 声明交换机 */channel.exchangeDeclare(CONFIRM_EXCHANGE2, BuiltinExchangeType.FANOUT);/* 声明队列 */channel.queueDeclare(CONFIRM_QUEUE2, true, false, false, null); System.out.println(String.format(声明交换机【%s】队列【%s】成功, CONFIRM_EXCHANGE2, CONFIRM_QUEUE2));/* 把队列绑定到交换机 */channel.queueBind(CONFIRM_QUEUE2, CONFIRM_EXCHANGE2, CONFIRM_ROUTE2);/* 关闭信道和连接 */channel.close();conn.close(); }
} 【2】生产者基于事务实现消息确认
/*** 消息确认生产者*/
public class ConfirmProducer {public static void main(String[] args) throws Exception {/* 获取连接*/Connection conn RBConnectionUtil.getConn();// 创建信道 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(); // 开启事务int i0;for (String msg : messages) {try {send(channel, msg); // 发送消息 System.out.println(msg is sent);if (i 5) {int temp 1/0; // 抛出0除异常 } channel.txCommit(); // 提交事务 System.out.println(消息发送完成事务提交消息 msg); } catch(Exception e) { channel.txRollback(); // 抛出异常 回滚事务System.out.println(抛出异常消息发送取消事务回滚消息 msg); }}channel.close(); conn.close();}private static void send(Channel channel, String msg) throws Exception { // 发送消息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
消息发送完成事务提交消息first.-04130828
second..-04130828 is sent
消息发送完成事务提交消息second..-04130828
third...-04130828 is sent
消息发送完成事务提交消息third...-04130828
fourth....-04130828 is sent
消息发送完成事务提交消息fourth....-04130828
fiveth.....-04130828 is sent
消息发送完成事务提交消息fiveth.....-04130828
6th.....-04130828 is sent
抛出异常消息发送取消事务回滚消息6th.....-04130828
7th.....-04130828 is sent
抛出异常消息发送取消事务回滚消息7th.....-04130828
8th.....-04130828 is sent
抛出异常消息发送取消事务回滚消息8th.....-04130828
9th.....-04130828 is sent
抛出异常消息发送取消事务回滚消息9th.....-04130828
10th.....-04130828 is sent
抛出异常消息发送取消事务回滚消息10th.....-04130828【4】rabbitmq首页查看队列中的消息总数
