RabbitMQ事件消息实现原理 - IT评测·应用市场-qidao123.com

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
public class RabbitMQTransactionExample {
// 定义队列名称
private final static String QUEUE_NAME = "transaction_queue";
public static void main(String[] args) {
// 1. 创建连接工厂并配置参数（请确保本地已启动 RabbitMQ 服务）
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
// 如有需要，可设置用户名和密码（默认为 guest/guest）
// factory.setUsername("guest");
// factory.setPassword("guest");
Connection connection = null;
Channel channel = null;
try {
// 2. 建立连接和创建通道
connection = factory.newConnection();
channel = connection.createChannel();
// 3. 声明一个持久化队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
System.out.println("队列声明成功: " + QUEUE_NAME);
// 4. 开启事务模式
channel.txSelect();
System.out.println("事务模式已开启（tx.select）");
// 5. 构造消息内容
String message = "Hello, RabbitMQ事务消息！";
System.out.println("准备发布消息: " + message);
// 6. 发布消息到队列中
// 使用 MessageProperties.PERSISTENT_TEXT_PLAIN 设置消息为持久化
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
System.out.println("消息已在事务中发布");
// 7. 模拟业务逻辑判断：根据条件决定提交或回滚事务
// 例如，可以根据某些校验结果来判断是否继续提交
boolean condition = true; // 修改为 false 可模拟回滚场景
if (condition) {
// 提交事务，将所有事务内的消息写入队列
channel.txCommit();
System.out.println("事务已提交（tx.commit），消息投递成功");
} else {
// 回滚事务，撤销本次消息操作
channel.txRollback();
System.out.println("事务已回滚（tx.rollback），消息未投递");
}
} catch (Exception e) {
System.err.println("发生异常: " + e.getMessage());
e.printStackTrace();
// 出现异常时，尝试回滚事务
if (channel != null) {
try {
channel.txRollback();
System.out.println("因异常回滚事务（tx.rollback）");
} catch (Exception ex) {
System.err.println("回滚失败: " + ex.getMessage());
}
}
} finally {
// 8. 关闭通道和连接，释放资源
if (channel != null) {
try {
channel.close();
} catch (Exception e) {
// 忽略关闭异常
}
}
if (connection != null) {
try {
connection.close();
} catch (Exception e) {
// 忽略关闭异常
}
}
}
}
}

复制代码