RocketMQ生产消费指南

@Component
public class DemoLocalTransactionChecker implements LocalTransactionChecker {
    @Override
    public TransactionStatus check(Message msg) {
        System.out.println("开始回查本地事务状态");
        return TransactionStatus.CommitTransaction; //根据本地事务状态检查结果返回不同的TransactionStatus
    }
}

@Configuration
public class TransactionProducerClient {

    @Autowired
    private MqConfig mqConfig;

    @Autowired
    private DemoLocalTransactionChecker localTransactionChecker;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public TransactionProducerBean buildTransactionProducer() {
        TransactionProducerBean producer = new TransactionProducerBean();
        producer.setProperties(mqConfig.getMqPropertie());
        producer.setLocalTransactionChecker(localTransactionChecker);
        return producer;
    }

}

public void txMessage() {
    Message message = new Message("oak-test-1","tag1","hello rocketmq! Transaction Message~".getBytes(StandardCharsets.UTF_8));
    try {
        SendResult sendResult = buildTransactionProducer.send(message, (msg, arg) -> {
            // 消息ID（有可能消息体一样，但消息ID不一样，当前消息属于半事务消息，所以消息ID在消息队列RocketMQ版控制台无法查询）。
            String msgId = msg.getMsgID();
            // 消息体内容进行crc32，也可以使用其它的如MD5。
            long crc32Id = 666L;
            // 消息ID和crc32id主要是用来防止消息重复。
            // 如果业务本身是幂等的，可以忽略，否则需要利用msgId或crc32Id来做幂等。
            // 如果要求消息绝对不重复，推荐做法是对消息体使用crc32或MD5来防止重复消息。

            TransactionStatus transactionStatus = TransactionStatus.Unknow;
            try {
                boolean isCommit = true; // 这里的true由业务代码返回

                if (isCommit) {
                    // 本地事务已成功则提交消息。
                    transactionStatus = TransactionStatus.CommitTransaction;
                } else {
                    // 本地事务已失败则回滚消息。
                    transactionStatus = TransactionStatus.RollbackTransaction;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println(msg.getMsgID());
            return transactionStatus;

        }, null);
    } catch (Exception e) {
        // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。
        System.out.println(new Date() + " Send mq message failed. Topic is:" + message.getTopic());
        e.printStackTrace();
    }
}

@Component
public class DemoMessageListener implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {

        System.out.println("Receive: " + message);
        try {
            System.out.print(new String(message.getBody(), StandardCharsets.UTF_8));
            return Action.CommitMessage;
        } catch (Exception e) {
            //消费失败
            return Action.ReconsumeLater;
        }
    }
}

@Bean(initMethod = "start", destroyMethod = "shutdown")
public ConsumerBean buildConsumer() {
    ConsumerBean consumerBean = new ConsumerBean();
    //配置文件
    Properties properties = mqConfig.getMqPropertie();
    properties.setProperty(PropertyKeyConst.GROUP_ID,mqConfig.getGroupId());
    //将消费者线程数固定为20个 20为默认值
    properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
    // 订阅方式有两种集群订阅和广播订阅
    // 集群订阅:一个topic9条消息 消费者组中有3个消费者 那么每个消费者平均消费3条消息
    // 广播订阅: 一个topic9条消息 消费者组中有3个消费者 那么每个消费者都会消费9条消息
    // 集群订阅方式设置（不设置的情况下，默认为集群订阅方式）。 订单这种肯定选择集群订阅啦
    properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
    consumerBean.setProperties(properties);
    //订阅关系
    Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
    Subscription subscription = new Subscription();
    subscription.setTopic(mqConfig.getTopic());
    subscription.setExpression(mqConfig.getTag());
    subscriptionTable.put(subscription, messageListener);
    //订阅多个topic如上面设置

    consumerBean.setSubscriptionTable(subscriptionTable);
    return consumerBean;
}