RabbitMQ之事务机制

事务相关理论回顾
1：什么是事务
事务要么成功 要么失败 传统事务有4个主要特性:原子性、一致性、隔离性、持久性
2：微服务带来挑战-分布式事务
◆ 在传统单体应用中，事务在本地即可完成
◆ 随着后端架构的微服务化，事务无法在本地完成
◆ 所以需要将事务“分布式化”
保证事务的前提
分布式框架理论 ACID
事务正确执行的四个基本要素
◆ 原子性(Atomicity)
◆ 一致性(Consistency)
◆ 隔离性(Isolation)
◆ 持久性(Durability)

分布式框架理论 CAP
一致性、可用性、分区容忍性不可能三者兼顾
◆ 一致性(Consistency)
◆ 可用性(Availability)
◆ 分区容忍性 (Partition tolerance)

分布式框架理论 BASE
由于CAP无法同时满足，基于I程实际，提出了BASE理论
◆ Basically Available (基本可用)
◆ Soft state (软状态)
◆ Eventually consistent (最终一致性)
分布式事务框架设计根据上述分析，分布式事务框架应该包含以下部分
◆ 发送失败重试
◆ 消费失败重试
◆ 死信告警
以下为代码实战部分

=================>配置类
@Configuration
public class DirectExchangeConfiguration {
    /**
     * 创建一个 Queue
     *
     * @return Queue
     */
    @Bean
    public Queue queue07() {
        // Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除
        return new Queue(
                Message07.QUEUE,
                true,
                false,
                false);
    }

    /**
     * 创建 Direct Exchange
     *
     * @return DirectExchange
     */
    @Bean
    public DirectExchange exchange07() {
        // name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它
        return new DirectExchange(Message07.EXCHANGE,
                true,
                false);
    }

    /**
     * 创建 Binding
     * Exchange：Message07.EXCHANGE
     * Routing key：Message07.ROUTING_KEY
     * Queue：Message07.QUEUE
     *
     * @return Binding
     */
    @Bean
    public Binding binding07() {
        return BindingBuilder
                .bind(queue07()).to(exchange07())
                .with(Message07.ROUTING_KEY);
    }
@Configuration
public class RabbitConfiguration {
    @Resource
    ConnectionFactory connectionFactory;
    @Resource
    RabbitTemplate rabbitTemplate;

    /**
     * 注入一个事务管理器 RabbitTransactionManager
     *
     * @return RabbitTransactionManager
     */
    @Bean
    public RabbitTransactionManager rabbitTransactionManager() {
        // 设置 RabbitTemplate 支持事务
        rabbitTemplate.setChannelTransacted(true);

        // 创建 RabbitTransactionManager 对象
        return new RabbitTransactionManager(connectionFactory);
    }
}
=================》生产者
@Component
@Slf4j
public class Producer07 {
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 在发送消息方法上，我们添加了 @Transactional 注解，声明事务。
     * 因为我们创建了 RabbitTransactionManager 事务管理器，所以这里会创建 RabbitMQ 事务
     * <p>
     * 当然也可以使用编程式事务
     * channel.txSelect();
     * channel.basicPublish();
     * channel.txCommit();
     * channel.txRollback(); // 回滚事务
     *
     * @param id         id
     * @param routingKey routingKey
     * @throws InterruptedException 异常
     */
    @Transactional(rollbackFor = Exception.class)
    public void syncSend(String id, String routingKey) throws InterruptedException {
        // 创建 Message07 消息
        Message07 message = new Message07();
        message.setId(id);
        // 同步发送消息
        rabbitTemplate.convertAndSend(Message07.EXCHANGE, routingKey, message);
        log.info("[{}][Producer07 syncSend][此时已经发送][id:{}]", LocalDateTime.now(), id);
        /*
            睡上 10s 方便看效果
            如果同步发送消息成功后，Consumer 立即消费到该消息，说明未生效
            如果 Consumer 是 10 秒之后，才消费到该消息，说明已生效
         */
        TimeUnit.SECONDS.sleep(10);
    }
==========================》消费者

@Component
@RabbitListener(queues = Message07.QUEUE)
@Slf4j
public class Consumer07 {

    @RabbitHandler
    public void onMessage(Message07 message) {
        log.info("[{}][Consumer07 onMessage][线程编号:{} 消息内容：{}]", LocalDateTime.now(), Thread.currentThread().getId(), message);
    }
}
==================================>测试类
 @Test
    void syncSend() throws InterruptedException {
        String id = UUID.randomUUID().toString();
        producer07.syncSend(id, Message07.ROUTING_KEY);
        log.info("[{}][test producer07 syncSend][id:{}] 发送成功", LocalDateTime.now(), id);

        TimeUnit.SECONDS.sleep(2);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

以上的是RabbitMQ之事务机制实现的代码 若不了解rabbitmq的基本使用 建议先看看我前面对应的文章 文章链接：点我—>let’s go
若需完整代码 可识别二维码后 给您发代码。