RocketMQ 源码分析 —— 集成 Spring Boot

<!-- pom -->

<modules>
    <module>rocketmq-spring-boot-parent</module>
    <module>rocketmq-spring-boot</module>
    <module>rocketmq-spring-boot-starter</module>
    <!-- Note: The samples need to mvn compiple in its own directory
            <module>rocketmq-spring-boot-samples</module>
    -->
    <module>rocketmq-spring-boot-samples</module>
</modules>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-parent</artifactId>
        <version>2.0.2-SNAPSHOT</version>
        <relativePath>../rocketmq-spring-boot-parent/pom.xml</relativePath>
    </parent>

    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <packaging>jar</packaging>

    <name>RocketMQ Spring Boot Starter</name>
    <description>SRocketMQ Spring Boot Starter</description>
    <url>https://github.com/apache/rocketmq-spring</url>

    <dependencies>
        <!-- Spring Boot RocketMQ 具体实现 -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot</artifactId>
        </dependency>
        <!-- Spring Boot Starter -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <!-- 提供 Validation 功能 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-validation</artifactId>
        </dependency>
    </dependencies>
    
</project>

// RocketMQMessageListener.java

@Target(ElementType.TYPE) // 表名，注解在类上
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RocketMQMessageListener {

    /**
     * 消费分组
     *
     * Consumers of the same role is required to have exactly same subscriptions and consumerGroup to correctly achieve
     * load balance. It's required and needs to be globally unique.
     *
     *
     * See <a href="http://rocketmq.apache.org/docs/core-concept/">here</a> for further discussion.
     */
    String consumerGroup();

    /**
     * 消费主体
     *
     * Topic name.
     */
    String topic();

    /**
     * 选择消息的方式
     *
     * Control how to selector message.
     *
     * @see SelectorType
     */
    SelectorType selectorType() default SelectorType.TAG;

    /**
     * 选择消息的表达式
     *
     * Control which message can be select. Grammar please see {@link SelectorType#TAG} and {@link SelectorType#SQL92}
     */
    String selectorExpression() default "*";

    /**
     * 消费模式
     *
     * Control consume mode, you can choice receive message concurrently or orderly.
     */
    ConsumeMode consumeMode() default ConsumeMode.CONCURRENTLY;

    /**
     * 消费模型
     *
     * Control message mode, if you want all subscribers receive message all message, broadcasting is a good choice.
     */
    MessageModel messageModel() default MessageModel.CLUSTERING;

    /**
     * 消费线程数
     *
     * Max consumer thread number.
     */
    int consumeThreadMax() default 64;

}

// RocketMQTransactionListener.java

@Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE})  // 表名，注解在类上
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Component // 默认带了 @Component 注解，所以只要添加到了类上，就会注册成 Spring Bean 对象
public @interface RocketMQTransactionListener {

    /**
     * 生产者分组
     *
     * Declare the txProducerGroup that is used to relate callback event to the listener, rocketMQTemplate must send a
     * transactional message with the declared txProducerGroup.
     * <p>
     * <p>It is suggested to use the default txProducerGroup if your system only needs to define a TransactionListener class.
     */
    String txProducerGroup() default RocketMQConfigUtils.ROCKETMQ_TRANSACTION_DEFAULT_GLOBAL_NAME;

    /**
     * Set ExecutorService params -- corePoolSize
     */
    int corePoolSize() default 1;

    /**
     * Set ExecutorService params -- maximumPoolSize
     */
    int maximumPoolSize() default 1;

    /**
     * Set ExecutorService params -- keepAliveTime
     */
    long keepAliveTime() default 1000 * 60; //60ms

    /**
     * Set ExecutorService params -- blockingQueueSize
     */
    int blockingQueueSize() default 2000;

}

// RocketMQConfigUtils.java

public static final String ROCKETMQ_TRANSACTION_DEFAULT_GLOBAL_NAME = "rocketmq_transaction_default_global_name";

// RocketMQProperties.java

@ConfigurationProperties(prefix = "rocketmq") // 配置文件中 rocketmq 前缀
public class RocketMQProperties {

    /**
     * The name server for rocketMQ, formats: `host:port;host:port`.
     *
     * Namesrv 地址
     */
    private String nameServer;

    /**
     * Producer 配置
     */
    private Producer producer;

    // ... 省略 setting/getting 方法

    public static class Producer {

        /**
         * Name of producer.
         */
        private String group;

        /**
         * Millis of send message timeout.
         */
        private int sendMessageTimeout = 3000;

        /**
         * Compress message body threshold, namely, message body larger than 4k will be compressed on default.
         */
        private int compressMessageBodyThreshold = 1024 * 4;

        /**
         * Maximum number of retry to perform internally before claiming sending failure in synchronous mode.
         * This may potentially cause message duplication which is up to application developers to resolve.
         */
        private int retryTimesWhenSendFailed = 2;

        /**
         * <p> Maximum number of retry to perform internally before claiming sending failure in asynchronous mode. </p>
         * This may potentially cause message duplication which is up to application developers to resolve.
         */
        private int retryTimesWhenSendAsyncFailed = 2;

        /**
         * Indicate whether to retry another broker on sending failure internally.
         */
        private boolean retryNextServer = false;

        /**
         * Maximum allowed message size in bytes.
         */
        private int maxMessageSize = 1024 * 1024 * 4;

        // ... 省略 setting/getting 方法
    }

}

// RocketMQAutoConfiguration.java

@Configuration // 标识是配置类
@EnableConfigurationProperties(RocketMQProperties.class) // 指定 RocketMQProperties 自动配置
@ConditionalOnClass({ MQAdmin.class, ObjectMapper.class }) // 要求有 MQAdmin、ObjectMapper 类
@ConditionalOnProperty(prefix = "rocketmq", value = "name-server") // 要求有 rocketmq 开头，且 name-server 的配置
@Import({ JacksonFallbackConfiguration.class, ListenerContainerConfiguration.class }) // 引入 JacksonFallbackConfiguration 和 ListenerContainerConfiguration 配置类
@AutoConfigureAfter(JacksonAutoConfiguration.class) // 在 JacksonAutoConfiguration 之后初始化
public class RocketMQAutoConfiguration {

    // ... 省略配置方法
    
}

// RocketMQAutoConfiguration.java

@Bean
@ConditionalOnMissingBean(DefaultMQProducer.class) // 不存在 DefaultMQProducer Bean 对象
@ConditionalOnProperty(prefix = "rocketmq", value = {"name-server", "producer.group"}) // 要求有 rocketmq 开头，且 name-server、producer.group 的配置
public DefaultMQProducer defaultMQProducer(RocketMQProperties rocketMQProperties) {
    // 校验配置
    RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
    String nameServer = rocketMQProperties.getNameServer();
    String groupName = producerConfig.getGroup();
    Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
    Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");

    // 创建 DefaultMQProducer 对象
    DefaultMQProducer producer = new DefaultMQProducer(groupName);
    // 将 RocketMQProperties.Producer 配置，设置到 producer 中
    producer.setNamesrvAddr(nameServer);
    producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
    producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
    producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
    producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
    producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
    producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());

    return producer;
}

// RocketMQAutoConfiguration.java

@Bean(destroyMethod = "destroy") // 声明了销毁时，调用 destroy 方法
@ConditionalOnBean(DefaultMQProducer.class) // 有 DefaultMQProducer Bean 的情况下
@ConditionalOnMissingBean(RocketMQTemplate.class) // 不存在 RocketMQTemplate Bean 对象
public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer, ObjectMapper rocketMQMessageObjectMapper) {
    // 创建 RocketMQTemplate 对象
    RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
    // 设置其属性
    rocketMQTemplate.setProducer(mqProducer);
    rocketMQTemplate.setObjectMapper(rocketMQMessageObjectMapper);
    return rocketMQTemplate;
}

// RocketMQAutoConfiguration.java

@Bean
@ConditionalOnBean(RocketMQTemplate.class) // 有 RocketMQTemplate Bean 的情况下
@ConditionalOnMissingBean(TransactionHandlerRegistry.class) // 不存在 TransactionHandlerRegistry Bean 对象
public TransactionHandlerRegistry transactionHandlerRegistry(RocketMQTemplate template) {
    // 创建 TransactionHandlerRegistry 对象
    return new TransactionHandlerRegistry(template);
}

// RocketMQAutoConfiguration.java

@Bean(name = RocketMQConfigUtils.ROCKETMQ_TRANSACTION_ANNOTATION_PROCESSOR_BEAN_NAME) // Bean 的名字
@ConditionalOnBean(TransactionHandlerRegistry.class) // 有 TransactionHandlerRegistry Bean 的情况下
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public static RocketMQTransactionAnnotationProcessor transactionAnnotationProcessor(TransactionHandlerRegistry transactionHandlerRegistry) {
    // 创建 RocketMQTransactionAnnotationProcessor 对象
    return new RocketMQTransactionAnnotationProcessor(transactionHandlerRegistry);
}

// RocketMQConfigUtils.java

/**
 * The bean name of the internally managed RocketMQ transaction annotation processor.
 */
public static final String ROCKETMQ_TRANSACTION_ANNOTATION_PROCESSOR_BEAN_NAME = "org.springframework.rocketmq.spring.starter.internalRocketMQTransAnnotationProcessor";

// JacksonFallbackConfiguration.java

import com.fasterxml.jackson.databind.ObjectMapper;

@Configuration
@ConditionalOnMissingBean(ObjectMapper.class) // 不存在 ObjectMapper Bean 时
class JacksonFallbackConfiguration {

    @Bean
    public ObjectMapper rocketMQMessageObjectMapper() {
        return new ObjectMapper();
    }

}

// ListenerContainerConfiguration.java

@Configuration
public class ListenerContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {

    private static final Logger log = LoggerFactory.getLogger(ListenerContainerConfiguration.class);

    private ConfigurableApplicationContext applicationContext;

    /**
     * 计数器，用于在 {@link #registerContainer(String, Object)} 方法中，创建 DefaultRocketMQListenerContainer Bean 时，生成 Bean 的名字。
     */
    private AtomicLong counter = new AtomicLong(0);

    private StandardEnvironment environment;

    private RocketMQProperties rocketMQProperties;

    private ObjectMapper objectMapper;

    public ListenerContainerConfiguration(ObjectMapper rocketMQMessageObjectMapper, StandardEnvironment environment, RocketMQProperties rocketMQProperties) {
        this.objectMapper = rocketMQMessageObjectMapper;
        this.environment = environment;
        this.rocketMQProperties = rocketMQProperties;
    }

    @Override // 实现自 ApplicationContextAware 接口
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }
    
}

// ListenerContainerConfiguration.java

@Override // 实现自 SmartInitializingSingleton 接口
public void afterSingletonsInstantiated() {
    // <1> 获得所有 @RocketMQMessageListener 注解的 Bean 们
    Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class);
    // 遍历 beans 数组，生成（注册）对应的 DefaultRocketMQListenerContainer Bean 对象。
    if (Objects.nonNull(beans)) {
        beans.forEach(this::registerContainer);
    }
}

// ListenerContainerConfiguration.java

private void registerContainer(String beanName, Object bean) {
    // <1.1> 获得 Bean 对应的 Class 类名。因为有可能被 AOP 代理过。
    Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
    // <1.2> 如果未实现 RocketMQListener 接口，直接抛出 IllegalStateException 异常。
    if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) {
        throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName());
    }
    // <1.3> 获得 @RocketMQMessageListener 注解
    RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
    // <1.4> 校验注解配置
    validate(annotation);

    // <2.1> 生成 DefaultRocketMQListenerContainer Bean 的名字
    String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(), counter.incrementAndGet());
    GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
    // <2.2> 创建 DefaultRocketMQListenerContainer Bean 对象，并注册到 Spring 容器中。
    genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
            () -> createRocketMQListenerContainer(bean, annotation));

    // <3.1> 从 Spring 容器中，获得刚注册的 DefaultRocketMQListenerContainer Bean 对象
    DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName, DefaultRocketMQListenerContainer.class);
    // <3.2> 如果未启动，则进行启动
    if (!container.isRunning()) {
        try {
            container.start();
        } catch (Exception e) {
            log.error("Started container failed. {}", container, e);
            throw new RuntimeException(e);
        }
    }

    // 打印日志
    log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
}

// TransactionHandler.java

class TransactionHandler {

    /**
     * {@link RocketMQTransactionListener#txProducerGroup()}
     */
    private String name;
    /**
     * {@link RocketMQTransactionListener#corePoolSize()}
     * {@link RocketMQTransactionListener#maximumPoolSize()}
     * {@link RocketMQTransactionListener#maximumPoolSize()}
     * {@link RocketMQTransactionListener#keepAliveTime()}
     * {@link RocketMQTransactionListener#blockingQueueSize()}
     */
    private ThreadPoolExecutor checkExecutor;

    /**
     * Bean 的名字
     */
    private String beanName;
    /**
     * 对应的 RocketMQLocalTransactionListener 对象
     */
    private RocketMQLocalTransactionListener bean;
    private BeanFactory beanFactory;
    
    public void setCheckExecutor(int corePoolSize, int maxPoolSize, long keepAliveTime, int blockingQueueSize) {
        // 创建 ThreadPoolExecutor 对象
        this.checkExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize,
            keepAliveTime, TimeUnit.MILLISECONDS,
            new LinkedBlockingDeque<>(blockingQueueSize));
    }

    // ... 省略 setting/getting 方法
}

// TransactionHandlerRegistry.java

public class TransactionHandlerRegistry implements DisposableBean {


    private RocketMQTemplate rocketMQTemplate;
    /**
     * {@link TransactionHandler#name} 的 集合
     */
    private final Set<String> listenerContainers = new ConcurrentSet<>();

    public TransactionHandlerRegistry(RocketMQTemplate template) {
        this.rocketMQTemplate = template;
    }

    @Override
    public void destroy() throws Exception {
        listenerContainers.clear();
    }

    public void registerTransactionHandler(TransactionHandler handler) throws MQClientException {
        // <1.1> 不能声明重复的 TransactionHandler
        if (listenerContainers.contains(handler.getName())) {
            throw new MQClientException(-1,
                String.format("The transaction name [%s] has been defined in TransactionListener [%s]", handler.getName(), handler.getBeanName()));
        }
        // <1.2> 添加到 listenerContainers 中
        listenerContainers.add(handler.getName());

        // <2> 创建并启动 TransactionMQProducer
        rocketMQTemplate.createAndStartTransactionMQProducer(handler.getName(), handler.getListener(), handler.getCheckExecutor());
    }

}

// RocketMQTransactionAnnotationProcessor.java

public class RocketMQTransactionAnnotationProcessor implements BeanPostProcessor, Ordered, BeanFactoryAware {

    private static final Logger log = LoggerFactory.getLogger(RocketMQTransactionAnnotationProcessor.class);

    private BeanFactory beanFactory;
    /**
     * 不处理的类的集合
     */
    private final Set<Class<?>> nonProcessedClasses = Collections.newSetFromMap(new ConcurrentHashMap<>(64));

    private TransactionHandlerRegistry transactionHandlerRegistry;

    public RocketMQTransactionAnnotationProcessor(TransactionHandlerRegistry transactionHandlerRegistry) {
        this.transactionHandlerRegistry = transactionHandlerRegistry;
    }

    @Override // 实现自 BeanFactoryAware 接口
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }

    @Override // 实现自 BeanPostProcessor 接口
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }
    
    @Override // 实现自 Ordered 接口
    public int getOrder() {
        return LOWEST_PRECEDENCE;
    }

}

// RocketMQTransactionAnnotationProcessor.java

@Override // 实现自 BeanPostProcessor 接口
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
    // <1.1> 如果 nonProcessedClasses 不存在
    if (!this.nonProcessedClasses.contains(bean.getClass())) {
        // <1.2> 获得 Bean 对应的 Class 类名。因为有可能被 AOP 代理过。
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        // <1.3> 添加到 nonProcessedClasses 中，表示后面不处理。
        this.nonProcessedClasses.add(bean.getClass());
        // <2.1> 获得 @RocketMQTransactionListener 注解
        RocketMQTransactionListener listener = AnnotationUtils.findAnnotation(targetClass, RocketMQTransactionListener.class);
        // <2.2> 如果无注解，则不进行任何逻辑
        if (listener == null) { // for quick search
            log.trace("No @RocketMQTransactionListener annotations found on bean type: {}", bean.getClass());
        } else {
            // <2.3> 如果有注解，则注册到 TransactionHandlerRegistry 中
            try {
                processTransactionListenerAnnotation(listener, bean);
            } catch (MQClientException e) {
                log.error("Failed to process annotation " + listener, e);
                throw new BeanCreationException("Failed to process annotation " + listener, e);
            }
        }
    }

    return bean;
}

// RocketMQTransactionAnnotationProcessor.java

private void processTransactionListenerAnnotation(RocketMQTransactionListener listener, Object bean) throws MQClientException {
    // <1.1> 校验 @RocketMQTransactionListener 非空
    if (transactionHandlerRegistry == null) {
        throw new MQClientException("Bad usage of @RocketMQTransactionListener, " + "the class must work with RocketMQTemplate", null);
    }
    // <1.2> 如果未实现 RocketMQLocalTransactionListener 接口，直接抛出 IllegalStateException 异常。
    if (!RocketMQLocalTransactionListener.class.isAssignableFrom(bean.getClass())) {
        throw new MQClientException("Bad usage of @RocketMQTransactionListener, " + "the class must implement interface RocketMQLocalTransactionListener", null);
    }

    // <2> 将 @RocketMQTransactionListener 注解，创建成 TransactionHandler 对象
    TransactionHandler transactionHandler = new TransactionHandler();
    transactionHandler.setBeanFactory(this.beanFactory);
    transactionHandler.setName(listener.txProducerGroup());
    transactionHandler.setBeanName(bean.getClass().getName());
    transactionHandler.setListener((RocketMQLocalTransactionListener) bean);
    transactionHandler.setCheckExecutor(listener.corePoolSize(), listener.maximumPoolSize(), listener.keepAliveTime(), listener.blockingQueueSize());

    // <3> 注册 TransactionHandler 到 transactionHandlerRegistry 中
    transactionHandlerRegistry.registerTransactionHandler(transactionHandler);
}

// RocketMQHeaders.java

/**
 * Represents the RocketMQ message protocol that is used during the data exchange.
 */
public class RocketMQHeaders {

    public static final String KEYS = "KEYS";
    public static final String TAGS = "TAGS";
    public static final String TOPIC = "TOPIC";
    public static final String MESSAGE_ID = "MESSAGE_ID";
    public static final String BORN_TIMESTAMP = "BORN_TIMESTAMP";
    public static final String BORN_HOST = "BORN_HOST";
    public static final String FLAG = "FLAG";
    public static final String QUEUE_ID = "QUEUE_ID";
    public static final String SYS_FLAG = "SYS_FLAG";
    public static final String TRANSACTION_ID = "TRANSACTION_ID";
    public static final String PROPERTIES = "PROPERTIES";

}

// RocketMQUtil.java

public static org.apache.rocketmq.common.message.Message convertToRocketMessage(ObjectMapper objectMapper, String charset,
    String destination, org.springframework.messaging.Message<?> message) {
    // 生成消息的 bytes 数组
    Object payloadObj = message.getPayload();
    byte[] payloads;
    if (payloadObj instanceof String) {
        payloads = ((String) payloadObj).getBytes(Charset.forName(charset));
    } else if (payloadObj instanceof byte[]) {
        payloads = (byte[]) message.getPayload();
    } else {
        try {
            // <X>
            String jsonObj = objectMapper.writeValueAsString(payloadObj);
            payloads = jsonObj.getBytes(Charset.forName(charset));
        } catch (Exception e) {
            throw new RuntimeException("convert to RocketMQ message failed.", e);
        }
    }

    // 获得 topic、tag 属性
    String[] tempArr = destination.split(":", 2);
    String topic = tempArr[0];
    String tags = "";
    if (tempArr.length > 1) {
        tags = tempArr[1];
    }

    // 创建 Message 对象，传入上述变量到其构造方法
    org.apache.rocketmq.common.message.Message rocketMsg = new org.apache.rocketmq.common.message.Message(topic, tags, payloads);

    // 如果 MessageHeaders 非空，逐个处理
    MessageHeaders headers = message.getHeaders();
    if (Objects.nonNull(headers) && !headers.isEmpty()) {
        // 设置 KEYS 属性
        Object keys = headers.get(RocketMQHeaders.KEYS);
        if (!StringUtils.isEmpty(keys)) { // if headers has 'KEYS', set rocketMQ message key
            rocketMsg.setKeys(keys.toString());
        }

        // 设置 FLAG 属性
        Object flagObj = headers.getOrDefault("FLAG", "0");
        int flag = 0;
        try {
            flag = Integer.parseInt(flagObj.toString());
        } catch (NumberFormatException e) {
            // Ignore it
            log.info("flag must be integer, flagObj:{}", flagObj);
        }
        rocketMsg.setFlag(flag);

        // 设置 WAIT 属性
        Object waitStoreMsgOkObj = headers.getOrDefault("WAIT_STORE_MSG_OK", "true");
        boolean waitStoreMsgOK = Boolean.TRUE.equals(waitStoreMsgOkObj);
        rocketMsg.setWaitStoreMsgOK(waitStoreMsgOK);

        // 设置 USERS_ 属性
        headers.entrySet().stream()
            .filter(entry -> !Objects.equals(entry.getKey(), RocketMQHeaders.KEYS)
                && !Objects.equals(entry.getKey(), "FLAG")
                && !Objects.equals(entry.getKey(), "WAIT_STORE_MSG_OK")) // exclude "KEYS", "FLAG", "WAIT_STORE_MSG_OK"
            .forEach(entry -> {
                rocketMsg.putUserProperty("USERS_" + entry.getKey(), String.valueOf(entry.getValue())); // add other properties with prefix "USERS_"
            });

    }

    return rocketMsg;
}

// RocketMQUtil.java

public static TransactionListener convert(RocketMQLocalTransactionListener listener) {
    return new TransactionListener() {

        @Override
        public LocalTransactionState executeLocalTransaction(Message message, Object obj) {
            // <1> 转换 RocketMQ Message 转换成 Spring Message 对象
            // <2> 回调 RocketMQLocalTransactionListener 监听器
            RocketMQLocalTransactionState state = listener.executeLocalTransaction(convertToSpringMessage(message), obj);
            // <3> 转换 Spring Boot RocketMQ RocketMQLocalTransactionState 事务状态，成 RocketMQ  LocalTransactionState 事务状态
            return convertLocalTransactionState(state);
        }

        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
            // <1> 转换 RocketMQ Message 转换成 Spring Message 对象
            // <2> 回调 RocketMQLocalTransactionListener 监听器
            RocketMQLocalTransactionState state = listener.checkLocalTransaction(convertToSpringMessage(messageExt));
            // <3> 转换 Spring Boot RocketMQ RocketMQLocalTransactionState 事务状态，成 RocketMQ  LocalTransactionState 事务状态
            return convertLocalTransactionState(state);
        }

    };
}

// RocketMQListenerContainer.java

public interface RocketMQListenerContainer extends DisposableBean {

    /**
     * Setup the message listener to use. Throws an {@link IllegalArgumentException} if that message listener type is
     * not supported.
     */
    void setupMessageListener(RocketMQListener<?> messageListener);

}

// DefaultRocketMQListenerContainer.java

public class DefaultRocketMQListenerContainer implements InitializingBean, RocketMQListenerContainer, SmartLifecycle {

    private final static Logger log = LoggerFactory.getLogger(DefaultRocketMQListenerContainer.class);

    private long suspendCurrentQueueTimeMillis = 1000;

    /**
     * Message consume retry strategy<br> -1,no retry,put into DLQ directly<br> 0,broker control retry frequency<br>
     * >0,client control retry frequency.
     */
    private int delayLevelWhenNextConsume = 0;

    private String nameServer;

    private String consumerGroup;

    private String topic;

    private int consumeThreadMax = 64;

    private String charset = "UTF-8";

    private ObjectMapper objectMapper;

    private RocketMQListener rocketMQListener;

    private RocketMQMessageListener rocketMQMessageListener;

    /**
     * DefaultMQPushConsumer 对象。
     *
     * 在 {@link #initRocketMQPushConsumer()} 方法中，进行创建
     */
    private DefaultMQPushConsumer consumer;

    /**
     * 消息类型
     */
    private Class messageType;

    /**
     * 是否在运行中
     */
    private boolean running;

    // The following properties came from @RocketMQMessageListener.
    private ConsumeMode consumeMode;
    private SelectorType selectorType;
    private String selectorExpression;
    private MessageModel messageModel;
    
    // 省略各种 setting/getting 方法
}

// DefaultRocketMQListenerContainer.java

@Override
public void afterPropertiesSet() throws Exception {
    // 初始化 DefaultMQPushConsumer 对象
    initRocketMQPushConsumer();
    
    // 获得 messageType 属性
    this.messageType = getMessageType();
    log.debug("RocketMQ messageType: {}", messageType.getName());
}

// DefaultRocketMQListenerContainer.java

private void initRocketMQPushConsumer() throws MQClientException {
    Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
    Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
    Assert.notNull(nameServer, "Property 'nameServer' is required");
    Assert.notNull(topic, "Property 'topic' is required");

    // <1> 创建 DefaultMQPushConsumer 对象
    consumer = new DefaultMQPushConsumer(consumerGroup);
    // <2.1> 设置其属性
    consumer.setNamesrvAddr(nameServer);
    consumer.setConsumeThreadMax(consumeThreadMax);
    if (consumeThreadMax < consumer.getConsumeThreadMin()) {
        consumer.setConsumeThreadMin(consumeThreadMax);
    }

    // <2.2> 设置 messageModel 属性
    switch (messageModel) {
        case BROADCASTING:
            consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
            break;
        case CLUSTERING:
            consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
            break;
        default:
            throw new IllegalArgumentException("Property 'messageModel' was wrong.");
    }

    // <2.3> 设置 selectorType 属性
    switch (selectorType) {
        case TAG:
            consumer.subscribe(topic, selectorExpression);
            break;
        case SQL92:
            consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
            break;
        default:
            throw new IllegalArgumentException("Property 'selectorType' was wrong.");
    }

    // <2.4> 设置 messageListener 属性
    switch (consumeMode) {
        case ORDERLY:
            consumer.setMessageListener(new DefaultMessageListenerOrderly());
            break;
        case CONCURRENTLY:
            consumer.setMessageListener(new DefaultMessageListenerConcurrently());
            break;
        default:
            throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
    }

    // <3> 如果实现了 RocketMQPushConsumerLifecycleListener 接口，则调用 prepareStart 方法，执行准备初始化的方法
    if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
        ((RocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumer);
    }

}

// DefaultRocketMQListenerContainer#DefaultMessageListenerConcurrently.java

public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {

    @SuppressWarnings("unchecked")
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt messageExt : msgs) {
            log.debug("received msg: {}", messageExt);
            try {
                long now = System.currentTimeMillis();
                // <1> 转换消息
                // <2> 执行消费
                rocketMQListener.onMessage(doConvertMessage(messageExt));
                // 打印日志
                long costTime = System.currentTimeMillis() - now;
                log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
            } catch (Exception e) {
                // <3.1> 发生异常，返回稍后再消费
                log.warn("consume message failed. messageExt:{}", messageExt, e);
                context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }

        // <3.2> 返回消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

// DefaultRocketMQListenerContainer#DefaultMessageListenerOrderly.java

public class DefaultMessageListenerOrderly implements MessageListenerOrderly {

    @SuppressWarnings("unchecked")
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        // 遍历 MessageExt 消息
        for (MessageExt messageExt : msgs) {
            log.debug("received msg: {}", messageExt);
            try {
                long now = System.currentTimeMillis();
                // 转换消息
                // 执行消费
                rocketMQListener.onMessage(doConvertMessage(messageExt));
                // 打印日志
                long costTime = System.currentTimeMillis() - now;
                log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
            } catch (Exception e) {
                // 发生异常，设置中断消费一会
                log.warn("consume message failed. messageExt:{}", messageExt, e);
                context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        }

        // 返回消费成功
        return ConsumeOrderlyStatus.SUCCESS;
    }
}

// DefaultRocketMQListenerContainer.java

@Override
public void start() {
    // 如果已经启动，则抛出 IllegalStateException 异常
    if (this.isRunning()) {
        throw new IllegalStateException("container already running. " + this.toString());
    }

    // 启动 DefaultMQPushConsumer
    try {
        consumer.start();
    } catch (MQClientException e) {
        throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
    }

    // 标记已经启动
    this.setRunning(true);

    log.info("running container: {}", this.toString());
}

@Override // 实现自 SmartLifecycle->Lifecycle 接口
public boolean isRunning() {
    return running;
}

private void setRunning(boolean running) {
    this.running = running;
}

@Override // 实现自 SmartLifecycle 接口
public boolean isAutoStartup() {
    return true;
}

@Override // 实现自 SmartLifecycle->Phased 接口
public int getPhase() {
    // Returning Integer.MAX_VALUE only suggests that
    // we will be the first bean to shutdown and last bean to start
    return Integer.MAX_VALUE;
}

// DefaultRocketMQListenerContainer.java

@Override
public void stop() {
    // 必须在运行中
    if (this.isRunning()) {
        // 关闭 DefaultMQPushConsumer
        if (Objects.nonNull(consumer)) {
            consumer.shutdown();
        }
        // 标记不在启动
        setRunning(false);
    }
}

@Override
public void stop(Runnable callback) {
    stop();
    callback.run();
}

// DefaultRocketMQListenerContainer.java

@Override
public void destroy() {
    // 标记已经停止
    this.setRunning(false);
    // 关闭 DefaultMQPushConsumer
    if (Objects.nonNull(consumer)) {
        consumer.shutdown();
    }
    log.info("container destroyed, {}", this.toString());
}

// RocketMQListener.java

public interface RocketMQListener<T> { // <T> 泛型，声明消费的消息类型

    /**
     * 消费消息
     *
     * @param message 消息
     */
    void onMessage(T message);

}

// RocketMQPushConsumerLifecycleListener.java

public interface RocketMQPushConsumerLifecycleListener extends RocketMQConsumerLifecycleListener<DefaultMQPushConsumer> {
}

// RocketMQLocalTransactionListener.java

public interface RocketMQLocalTransactionListener {

    /**
     * 执行本地事务
     *
     * @param msg 消息
     * @param arg 方法参数
     * @return 本地事务状态
     */
    RocketMQLocalTransactionState executeLocalTransaction(final Message msg, final Object arg);

    /**
     * 检查本地事务状态
     *
     * @param msg 消息
     * @return 本地事务状态
     */
    RocketMQLocalTransactionState checkLocalTransaction(final Message msg);

}

// RocketMQTemplate.java

private static final  Logger log = LoggerFactory.getLogger(RocketMQTemplate.class);

/**
 * 消息生产者
 */
private DefaultMQProducer producer;

private ObjectMapper objectMapper;

private String charset = "UTF-8";

/**
 * 消息队列选择器
 */
private MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();

/**
 * TransactionMQProducer 的映射
 *
 * KEY：{@link TransactionMQProducer#getProducerGroup()} 事务生产者对应的分组
 */
private final Map<String, TransactionMQProducer> cache = new ConcurrentHashMap<>(); //only put TransactionMQProducer by now!!!

// RocketMQTemplate.java

@Override
public void afterPropertiesSet() throws Exception {
    Assert.notNull(producer, "Property 'producer' is required");
    producer.start();
}

// RocketMQTemplate.java

/**
 * <p> Send message in synchronous mode. This method returns only when the sending procedure totally completes.
 * Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS
 * notification, SMS marketing system, etc.. </p>
 * <p>
 * <strong>Warn:</strong> this method has internal retry-mechanism, that is, internal implementation will retry
 * {@link DefaultMQProducer#getRetryTimesWhenSendFailed} times before claiming failure. As a result, multiple
 * messages may potentially delivered to broker(s). It's up to the application developers to resolve potential
 * duplication issue.
 *
 * @param destination formats: `topicName:tags`
 * @param message     {@link org.springframework.messaging.Message}
 * @return {@link SendResult}
 */
public SendResult syncSend(String destination, Message<?> message) {
    return syncSend(destination, message, producer.getSendMsgTimeout());
}

public SendResult syncSend(String destination, Message<?> message, long timeout) {
    return syncSend(destination, message, timeout, 0);
}

public SendResult syncSend(String destination, Message<?> message, long timeout, int delayLevel) {
    // <1> 校验消息
    if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
        log.error("syncSend failed. destination:{}, message is null ", destination);
        throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
    }

    try {
        long now = System.currentTimeMillis();
        // <2> 将 message 转换成 RocketMQ Message 对象
        org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper, charset, destination, message);
        // <3> 设置 delayLevel 属性
        if (delayLevel > 0) {
            rocketMsg.setDelayTimeLevel(delayLevel);
        }
        // <4> 同步发送消息
        SendResult sendResult = producer.send(rocketMsg, timeout);
        // 打印日志
        long costTime = System.currentTimeMillis() - now;
        log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
        return sendResult;
    } catch (Exception e) {
        log.error("syncSend failed. destination:{}, message:{} ", destination, message);
        throw new MessagingException(e.getMessage(), e);
    }
}

// RocketMQTemplate.java

@Override // 实现自 DisposableBean 接口
public void destroy() {
    // 关闭 producer
    if (Objects.nonNull(producer)) {
        producer.shutdown();
    }

    // 关闭 cache
    for (Map.Entry<String, TransactionMQProducer> kv : cache.entrySet()) {
        if (Objects.nonNull(kv.getValue())) {
            kv.getValue().shutdown();
        }
    }
    cache.clear();
}

// RocketMQTemplate.java

/**
 * Create and start a transaction MQProducer, this new producer is cached in memory.
 * <p>Note: This method is invoked internally when processing {@code @RocketMQLocalTransactionListener}, it is not
 * recommended to directly use this method by user.
 *
 * @param txProducerGroup     Producer (group) name, unique for each producer
 * @param transactionListener TransactoinListener impl class
 * @param executorService     Nullable.
 * @return true if producer is created and started; false if the named producer already exists in cache.
 * @throws MessagingException
 */
public boolean createAndStartTransactionMQProducer(String txProducerGroup, RocketMQLocalTransactionListener transactionListener,  ExecutorService executorService) throws MessagingException {
    // <1> 如果已经存在，则直接返回
    txProducerGroup = getTxProducerGroupName(txProducerGroup);
    if (cache.containsKey(txProducerGroup)) {
        log.info(String.format("get TransactionMQProducer '%s' from cache", txProducerGroup));
        return false;
    }

    // <2> 创建 TransactionMQProducer 对象
    TransactionMQProducer txProducer = createTransactionMQProducer(txProducerGroup, transactionListener, executorService);
    try {
        // <3> 启动 TransactionMQProducer 对象
        txProducer.start();
        // <4> 添加到 cache 中
        cache.put(txProducerGroup, txProducer);
    } catch (MQClientException e) {
        throw RocketMQUtil.convert(e);
    }

    return true;
}

// RocketMQTemplate.java

/**
 * Send Spring Message in Transaction
 *
 * @param txProducerGroup the validate txProducerGroup name, set null if using the default name
 * @param destination     destination formats: `topicName:tags`
 * @param message         message {@link org.springframework.messaging.Message}
 * @param arg             ext arg
 * @return TransactionSendResult
 * @throws MessagingException
 */
public TransactionSendResult sendMessageInTransaction(final String txProducerGroup, final String destination, final Message<?> message, final Object arg) throws MessagingException {
    try {
        // <1> 获得 TransactionMQProducer 对象
        TransactionMQProducer txProducer = this.stageMQProducer(txProducerGroup);
        // <2> 将 message 转换成 RocketMQ Message 对象
        org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper, charset, destination, message);
        // <3> 发送事务消息
        return txProducer.sendMessageInTransaction(rocketMsg, arg);
    } catch (MQClientException e) {
        throw RocketMQUtil.convert(e);
    }
}