跟我学RocketMQ之批量消息发送源码解析

发表于 2019-08-08

上篇文章跟我学RocketMQ之消息发送源码解析中，我们已经对普通消息的发送流程进行了详细的解释，但是由于篇幅问题没有展开讲解批量消息的发送。本文中，我们就一起来集中分析一下批量消息的发送是怎样的逻辑。

DefaultProducer.send

RocketMQ提供了批量发送消息的API，同样在DefaultProducer.java中

@Override
public SendResult send(
    Collection<Message> msgs) throws MQClientException, RemotingException, 
        MQBrokerException, InterruptedException {
    return this.defaultMQProducerImpl.send(batch(msgs));
}

它的参数为Message集合，也就是一批消息。它的另外一个重载方法提供了发送超时时间参数

@Override
public SendResult send(Collection<Message> msgs,
    long timeout) throws MQClientException, RemotingException,
         MQBrokerException, InterruptedException {
    return this.defaultMQProducerImpl.send(batch(msgs), timeout);
}

可以看到是将消息通过batch()方法打包为单条消息，我们看一下batch方法的逻辑

自己写分布式事务框架之实现事务回滚

By SnoWalker

发表于 2019-08-08

上一篇文章中，我们对事务提交部分的框架逻辑及代码实现做了较为详细的讲解；本文中，我们接着分析一下事务回滚阶段的机理及代码实现逻辑。

这里主要看图的下半部分。

当事务下游应用达到最大消费次数，事务回滚被消息持久化之后，ShieldTXC的消息发送线程sendTxcMessageScheduler会扫描到待发送的回滚消息并投递到 [事务回滚队列]。

第一阶段：实现回滚逻辑

事务上游应用在启动过程中初始化了ShieldTxcConsumerListenerAdapter消费适配器，并通过ShieldTxcRollbackListener实现了回滚逻辑。

@Service
public class TxConsumeService implements InitializingBean {

    @Value("${shield.event.rocketmq.nameSrvAddr}")
    String nameSerAddr;
    @Value("${shield.event.rocketmq.topicSource}")
    String topic;

    @Override
    public void afterPropertiesSet() throws Exception {
        new ShieldTxcConsumerListenerAdapter(nameSerAddr, topic, new ShieldTxcRollbackListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println("测试消费【回滚】ShieldTxcRollbackListener");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        }));
    }
}

这段代码在之前的事务提交阶段已经讲解过，事务发起端需要自行实现回滚逻辑，这样才能在异常发生时与事务下游保持数据一致性。更多的细节此处就不再赘述。

自己写分布式事务框架之实现事务提交

By SnoWalker

发表于 2019-08-07

从本文开始，我们正式进入自己写分布式事务框架的部分。

笔者将这个本地消息表分布式事务框架命名为 shieldTXC，意思是 神盾事务框架，框架内核及demo案例的代码已经打包上传至github上，

地址为：shieldTXC源码地址。如果觉得这个喜欢可以点个star支持下。

题外话不多说，接下来我们一边看框架的原理图，在宏观上对框架的机理做一个了解，一边对相应的原理进行代码实现讲解，这样理论与实战相结合，相信会加深读者朋友的理解。

框架核心原理

首先看下框架的核心原理图。

看起来还是比较整洁的，这也是笔者写代码的一个宗旨：

好的框架可以复杂，但架构一定是优雅的、清晰的。

跟我学RocketMQ之消息发送源码解析

By SnoWalker

发表于 2019-08-07

本文我将带领读者朋友对RocketMQ生产者如何发送消息这一流程进行源码层面的解析。内容偏干，请自备白开水。

生产者初始化

进行消息发送的前提是先对生产者进行初始化，一段较为常规的生产者初始化示例代码如下

@Value("${rocketmq.nameServer}")
String nameSrvAddr;

@PostConstruct
public void init() {

    DefaultMQProducer defaultMQProducer =
            new DefaultMQProducer("PRODUCER_GROUP", true);
    defaultMQProducer.setNamesrvAddr(nameSrvAddr);
    // 发送失败重试次数
    defaultMQProducer.setRetryTimesWhenSendFailed(3);
    try {
        defaultMQProducer.start();
    } catch (MQClientException e) {
        throw new RuntimeException("Producer加载异常!", e);
    }
}