服务粉丝

我们一直在努力
当前位置：首页 > 财经 >

RocketMQ高级功能代码实现｜文末赠书

日期： 来源：程序猿DD收集编辑：
本文章我们将通过一些实际的案例，引出解决方案，并实际通过代码来实现。通过本节的学习，你可以确切的掌握实际编码能力。

本文选自北京大学出版社的《分布式中间件核心原理与RocketMQ最佳实践》一书

本书：分布式中间件核心原理与RocketMQ实战技术一本通：实战案例+操作步骤+执行效果图，手把手教你吃透分布式中间件技术，轻松实现从小白到大牛的职业跃迁。

1，事务消息代码实现

      之前我们已经在讨论订单业务消息丢失问题中引出了事务消息，本内容我们就实际用代码来实现一下事务消息吧。
    首先我们用原生代码来实现一下事务消息，下面是事务消息生产者TransactionProducer类的代码，具体代码解释已经用注释标明。
package com.huc.rocketmq.transaction;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.*;

/**
 * @author liumeng
 */
public class TransactionProducer {
public static void main(String[] args)
 throws MQClientException, UnsupportedEncodingException {
        // 这里是一个自定义的接收RocketMQ回调的监听接口
        TransactionListener transactionListener = new TransactionListenerImpl();
        // 创建支持事务消息的Producer,并指定生产者组
        TransactionMQProducer producer =
                new TransactionMQProducer("testTransactionGroup");
        // 指定一个线程池，用于处理RocketMQ回调请求的
        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("testThread");
                        return thread;
                    }
                }
        );
        // 给事务消息生产者设置线程池
        producer.setExecutorService(executorService);
        // 给事务消息生产者设置回调接口
        producer.setTransactionListener(transactionListener);
        // 启动生产者
        producer.start();
        // 构造一条订单支付成功的消息
        Message message = new Message(
                "PayOrderSuccessTopic",
                "testTag",
                "testKey",
                "订单支付消息".getBytes(RemotingHelper.DEFAULT_CHARSET)
        );

        // 将消息作为half消息发送出去
        try {
            TransactionSendResult result = producer.sendMessageInTransaction(message, null);
        } catch (Exception e) {
           // half消息发送失败
            // 订单系统执行回滚逻辑，比如退款、关闭订单
        }
    }
}
     针对于half消息发送失败的情况，是有可能一直接收不到消息发送失败的异常的，所以我们可以在发送half消息的时候，同时保存一份half消息到内存中，或者写入磁盘里，后台开启线程去检查half消息，如果超过10分钟都没有接到响应，就自动执行回滚逻辑。
      那么如果half消息成功了，如何执行本地事务逻辑呢？这就要说到代码中自定义的回调监听接口TransactionListenerImpl类了，代码如下：
package com.huc.rocketmq.transaction;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

public class TransactionListenerImpl implements TransactionListener {
    /**
     * 如果half消息发送成功了，就会回调这个方法，执行本地事务
     * @param message
     * @param o
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        // 执行订单本地业务，并根据结构返回commit/rollback
        try {
            // 本地事务执行成功，返回commit
            return LocalTransactionState.COMMIT_MESSAGE;
 }catch (Exception e){
            // 本地事务执行失败，返回rollback,作废half消息
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    /**
     * 如果没有正确返回commit或rollback，会执行此方法
     * @param messageExt
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        // 查询本地事务是否已经成功执行了,再次根据结果返回commit/rollback
        try {
            // 本地事务执行成功，返回commit
            return LocalTransactionState.COMMIT_MESSAGE;
        }catch (Exception e){
            // 本地事务执行失败，返回rollback,作废half消息
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
}

     到这里事务消息的代码我们就完成了，但是我相信小伙伴们不会满足于仅仅使用原生代码实现，那接下来我们就用Spring Boot重写编写一次相同的逻辑。

     使用Spring Boot项目后，我们还是先准备一个消息的实体类TranMessage，代码如下：

package com.huc.rocketmq.transaction.spring;

/**
 * 事务消息实体
 */
public class TranMessage {

    public static final String TOPIC = "Tran";

    /**
     * 编号
     */
    private Integer id;

    public TranMessage setId(Integer id) {
        this.id = id;
        return this;
    }

    public Integer getId() {
        return id;
    }
    @Override
public String toString() {
        return "TranMessage{" +
                "id=" + id +
                '}';
    }
}
 然后我们编写事务消息的生产者TranProducer：

package com.huc.rocketmq.transaction.spring;

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
public class TranProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public SendResult sendMessageInTransaction(Integer id) {
        // 创建TranMessage消息
        Message<TranMessage> message = MessageBuilder
                .withPayload(new TranMessage().setId(id)).build();
        // 发送事务消息
        return rocketMQTemplate.sendMessageInTransaction(TranMessage.TOPIC,
                message,id);
    }

}

同样的，我们需要编写一个回调监听的实现类，用于自定义处理本地事务，返回commit或者rollback消息。代码如下：

package com.huc.rocketmq.transaction.spring;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
// 注解中可以指定线程池参数
@RocketMQTransactionListener(corePoolSize=2,maximumPoolSize=5)
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行订单本地业务，并根据结构返回commit/rollback
        try {
            // 本地事务执行成功，返回commit
            return RocketMQLocalTransactionState.COMMIT;
        }catch (Exception e){
 // 本地事务执行失败，返回rollback,作废half消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 查询本地事务是否已经成功执行了,再次根据结果返回commit/rollback
        try {
            // 本地事务执行成功，返回commit
            return RocketMQLocalTransactionState.COMMIT;
        }catch (Exception e){
            // 本地事务执行失败，返回rollback,作废half消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}
     有了原生代码的实现经验，相信小伙伴们对于使用Spring Boot集成后的代码同样可以轻松看得懂。
      好了，至此事务消息的代码我们就已经实现了。

2，顺序消息代码实现

     有关消息乱序的出现原因以及解决方案我们已经在8.4.3小节中讲解过了，小伙伴们可以去复习一下，本节我们将直接讨论代码的实现，首先还是使用原生代码实现。

      经过之前的学习我们知道，解决消息乱序的方案就是把需要保证顺序的消息发送到同一个MessageQueue中，所以我们一定是需要编写一个MessageQueue的选择器的，RocketMQ的API中确实是有这部分内容的，就是MessageQueueSelector，下面就以原生代码异步的发送为例，在发送消息的时候指定队列选择器，主要代码如下，注释已经说明代码的含义：
producer.send(
        msg,
        new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Long orderId = (Long) arg; // 根据订单id选择发送的queue
                long index = orderId % mqs.size();// 用订单id于MessageQueue的数量取模
                return mqs.get((int) index); // 返回一个运算后固定的MessageQueue
            }
        },
        orderId, // 传入订单id
        new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }
            @Override
public void onException(Throwable throwable) {
                System.out.println(throwable);
            }
         }
 )；

     在发送消息时增加一个MessageQueueSelector，就可以实现统一订单id的消息一直会发送到同一个MessageQueue之中，可以解决消息乱序问题。

    接着我们来看消费者部分的代码实现，主要代码如下：

consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                       ConsumeOrderlyContext context) {
                try {
                    // 对有序的消息进行顺序处理
                    for (MessageExt t : msgs) {

                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                } catch (Exception e) {
                    // 如果消息处理出错，返回一个状态，暂停一会儿再来处理这批消息。
                    return 
ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
        });

      这里面要注意的是我们注册的监听器是MessageListenerOrderly，这个监听器为了保证顺序消费，Consumer会对每一个ConsumerQueue只使用一个线程来处理消息，如果使用了多线程，是无法避免消息乱序的。

     至此原生代码的实现已经完成了，Spring Boot的代码原理也是一样的。

     消息实体的代码我们就省略了，直接看生产者的代码，如下：

package com.huc.rocketmq.order.spring;

import com.huc.rocketmq.spring.DemoMessage;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class OrderProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public SendResult syncSend(Integer id) {
        // 创建 DemoMessage 消息
        DemoMessage message = new DemoMessage();
        message.setId(id);
        // 同步发送消息
        return rocketMQTemplate.syncSendOrderly(DemoMessage.TOPIC,
                message,String.valueOf(id));
 }

    public void asyncSend(Integer id, SendCallback callback) {
        // 创建 DemoMessage 消息
        DemoMessage message = new DemoMessage();
        message.setId(id);
        // 异步发送消息
        rocketMQTemplate.asyncSendOrderly(DemoMessage.TOPIC,
                message,String.valueOf(id),callback);
    }

    public void onewaySend(Integer id) {
        // 创建 DemoMessage 消息
        DemoMessage message = new DemoMessage();
        message.setId(id);
        // oneway 发送消息
        rocketMQTemplate.sendOneWayOrderly(DemoMessage.TOPIC, 
                message,String.valueOf(id));
    }

}

    以上代码中可以看出，每个发送方法中都调用了对应的Orderly方法，并传入了一个id值，默认根据id值采用SelectMessageQueueByHash策略来选择MessageQueue。

     接下来我们继续看消费者代码的实现。

package com.huc.rocketmq.order.spring;

import com.huc.rocketmq.spring.DemoMessage;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(
        topic = DemoMessage.TOPIC,
        consumerGroup = "demo-consumer-group-" + DemoMessage.TOPIC,
        consumeMode = ConsumeMode.ORDERLY // 设置为顺序消费
)
public class OrderConsumer implements RocketMQListener<DemoMessage> {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void onMessage(DemoMessage message) {
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

}

      可以看到消费者代码改动很小，只需要在@RocketMQMessageListener注解中新增consumeMode = ConsumeMode.ORDERLY，就可以指定顺序消费了，小伙伴们可以大胆的猜测它的实现原理，和我们的原生代码实现的方式是相同的。

3，消息过滤代码实现

     RocketMQ是包含消息过滤功能的，现在假如我们不使用消息过滤功能，获取到一个Topic中的消息可能包含了相关主题的多个表的信息。
      如果我们的需求是根据获取的消息同步某张表A的数据，那么就需要在获取消息后自行判断消息是否属于表A，如果属于表A才去处理，如果不是表A就直接丢弃。
     这种做法多了一层逻辑判断，自然会对系统的性能产生影响。这个时候RocketMQ的过滤机制就可以展示它的作用了，我们在发送消息的时候可以直接给消息指定tag和属性，主要代码如下：
// 构建消息对象
        Message msg = new Message(
                topic, //这里指定的是topic
                "A",//这里存放的Tag 消费者会根据tag进行消息过滤
                message.getBytes(RemotingHelper.DEFAULT_CHARSET));
        // 我们还可以设置一些用户自定义的属性
        msg.putUserProperty("name","value");

      消费者在消费数据时就可以根据tag和属性进行过滤了，比如下边的写法：

// 订阅test Topic , 第二个参数是通过tag过滤，意思是过滤出tag为A或B的消息
        consumer.subscribe("test", "A||B");

       对应到spring boot中的实现也很简单，生产者部分关键代码如下：

// 创建 DemoMessage 消息
        Message<DemoMessage> message = MessageBuilder
                .withPayload(new DemoMessage().setId(id))
                .setHeader(MessageConst.PROPERTY_TAGS,"A")// 设置消息的tag
                .build();

      消费者过滤的主要代码如下：

@RocketMQMessageListener(
        topic = DemoMessage.TOPIC,
        consumerGroup = "demo-consumer-group-" + DemoMessage.TOPIC,
        selectorExpression = "A||B" // 通过tag过滤
)

     消费者部分只要在@RocketMQMessageListener注解中增加selectorExpression属性就可以了。

4，延时消息代码实现

     在讨论延时消息的代码实现之前，先讨论一下电商系统的超时未支付业务流程。如图1所示：

图1  放弃支付流程

     这个流程的关键问题就是超时未支付的订单处于“待支付”状态，并锁定了库存，当时我们提出的解决方案就是提供一个后台线程，来扫描待支付订单，如果超过30分钟还未支付，就把订单关闭，解锁库存。
      小伙伴们可以思考一下，这样的解决方案真的可以在生产环境落地吗？
     首先，后台线程不停的扫描订单数据，如果订单数据量很大，就会导致严重的系统性能问题。
     其次，如果我们的订单系统是一个分布式系统，你的后台线程要如何部署？多久扫描一次？
      所以，使用后台线程扫描订单数据并不是一个优雅的解决方案，这个时候本小节的主人公延时消息就该出场了。
      RocketMQ的延时消息可以做到这样的效果，订单系统发送一条消息，等30分钟后，这条消息才可以被消费者消费。所以我们引入延时消息后，就可以单独准备一个订单扫描服务，来消费延时消息，当它获得消息的时候再去验证订单是否已经支付，如果已经支付什么都不用做，如果还未支付就去进行关闭订单，解锁库存的操作。如图2所示：

图2   延时消息放弃支付流程

     使用延时消息后，就可以避免扫描大量订单数据的操作了，而且订单扫描服务也可以分布式部署多个，只要同时订阅一个Topic就可以了。
       应用场景我们已经了解了，现在我们来看一下代码应该如何实现。
        延时消息使用原生代码实现特别容易，主要代码如下：
// 构建消息对象
        Message msg = new Message(
                topic, //这里指定延时消息的topic
                message.getBytes(RemotingHelper.DEFAULT_CHARSET));
        // 指定延时级别为3
        msg.setDelayTimeLevel(3);
        producer.send(msg);

      可以看到最核心的内容就是msg.setDelayTimeLevel(3)，设置了延迟级别。

      RocketMQ支持的延迟级别有18个，这个我们之前已经介绍过了，如下：

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

‍

      所以设置为3代表10s后消息可以被消费者消费。

      消费者的代码这里就不演示了，没有什么特殊的写法。

      下面我们来看一下Spring Boot的生产者代码实现：

// 创建 DemoMessage 消息
        Message<DemoMessage> message = MessageBuilder
                .withPayload(new DemoMessage().setId(id))
                .build();
        // 同步发送消息
        return rocketMQTemplate.syncSend(DemoMessage.TOPIC,
                message,
                30*1000,
                3);// 此处设置的就是延时级别

声明：本文选自北京大学出版社的《分布式中间件核心原理与RocketMQ最佳实践》一书，略有修改，经出版社授权刊登于此。

编辑推荐

1.易读易懂：没有高深的理论，每一章都是以实例为主，读者参考源码，修改实例，看得懂、学得会、做得出。
2.内容系统，实战性强。本书从零开始讲解，然后逐步深入相关分布式技术，从而达到从入门到精通的学习效果。
3.代码精彩，注释详细。深入剖析代码的底层逻辑，知其然更知其所以然。
4.配套资源：赠送全书案例源文件助力学习。

抽奖赠书

本次福利将送出《分布式中间件核心原理与RocketMQ最佳实践》* 5本，超高中奖率（参与人少，多期中奖率超过90%）

为避免撸羊毛的用户参与抽奖，让更多真实用户获得社区福利，现在把抽奖方式切换到 spring4all.com 网上通过积分参与。

本次抽奖地址：http://spring4all.com/forum-post/2112.html

还有一大波福利正在路上，一起来参与社区内容的建设，一起学习一起成长吧！
如何获取积分？日常登录spring4all.com，发帖交流、分享资源、回帖帮助他人等有益内容的，均可获得积分！

文章推荐

相关阅读

  • 俄方决定：只延长60天

  • 当地时间18日，俄罗斯外交部发言人扎哈罗娃在接受俄媒采访时表示，俄罗斯决定将黑海港口农产品外运协议只延长60天。俄方已通过口头和照会形式将上述决定通知有关各方。（更多国际

  • GitHub大数据：10%的程序员泄密

  • GitGuardian近日扫描了2022年GitHub的10亿次代码提交（较2021年增长20%），结果发现了1000万条机密信息，较2021年增长了67%。此外，调查结果还显示，10%的程序员（GitHub代码提交者）都泄露

  • 什么？脏话越多，代码越好！

  • 将 脚本之家 设为“星标⭐”第一时间收到文章更新来源 | 了不起的程序员（ID：great_developer）如若转载请联系原公众号前两天看到一则代码注释里出现各种脏话的消息，这让我想起了

  • 如何利用ChatGPT搞科研？

  • 转载自 知乎：芯片斯多葛量子位 | 公众号 QbitAI这位研究僧，GPT-4都发布了，你还在纯人工搞科研吗？已经有人实测：ChatGPT科研全流程都能用，尤其是对于英语非母语的科研人员，可以极大

  • 基辛格赞赏中国此举：“和平缔造者”！

  • 美国《华盛顿邮报》3月16日发表评论文章，作者伊格内修斯在文中提到了他本周对美国前国务卿基辛格的采访，其间基辛格就中国促成沙伊复交一事作出评论。文章称，基辛格对伊格内修

  • 逆向之 Sublime Text3破解

  • 关于Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Py

  • 知乎热议！如何利用ChatGPT搞科研？

  • 【做计算 找华算】理论计算助攻顶刊，10000+成功案例，全职海归技术团队、正版商业软件版权！经费预存选华算，高至15%预存增值！来源丨 QbitAI、量子位这位研究僧，GPT-4都发布了，你还在

聚合标签

中国 自己的 美国 都是 疫情 的人 的是 亿元 三星 手机 华为 这款 车型 万元 企业 日本 小米 俄罗斯 民警 项目

热门文章

  • “复活”半年后　京东拍拍二手杀入公益事业

  • 京东拍拍二手“复活”半年后，杀入公益事业，试图让企业捐的赠品、家庭闲置品变成实实在在的“爱心”。 把“闲置品”变爱心 6月12日，“益心一益·守护梦想每一步”2018年四

最新文章

  • RocketMQ高级功能代码实现｜文末赠书

  • 本文章我们将通过一些实际的案例，引出解决方案，并实际通过代码来实现。通过本节的学习，你可以确切的掌握实际编码能力。本文选自北京大学出版社的《分布式中间件核心原理与Rock

  • Java中如何解析、格式化、生成SQL语句？

  • 大家好，我是TJ一个励志推荐10000款开源项目与工具的程序员昨天在群里看到有小伙伴问，Java里如何解析SQL语句然后格式化SQL，是否有现成类库可以使用？之前TJ没有做过这类需求，所以