首页 > 基础资料博客日记

RocketMQ快速入门：集成java客户端实现各类消息发送｜异步、同步、顺序、单向、延迟、事务（五）附带源码

2024-09-21 13:00:07基础资料围观109次

文章RocketMQ快速入门：集成java客户端实现各类消息发送｜异步、同步、顺序、单向、延迟、事务（五）附带源码分享给大家，欢迎收藏Java资料网，专注分享技术知识

0. 引言

前面的章节中，我们已经针对rocketmq的基本概念和消息发送、消费流程进行了讲解，但实际在开发中如何实现rocketmq的接入、实现消息发送、消费还没有落实，那么今天，我们继续来学习如何基于java client集成rocketMQ

1. 集成rocketMQ

1、spring或其他java框架集成rocektmq很简单，只需要引入rocketmq依赖即可，这里的版本号与你的rocketmq保持一致

<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.8.0</version>
</dependency>

1.1 消息同步发送

官方文档：https://rocketmq.apache.org/zh/docs/4.x/producer/02message1/

1.1.1 实现步骤

1、实现消息发送需要利用生产者类DefaultMQProducer实现，以下示例简单消息的同步发送

public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        // 声明group
        DefaultMQProducer producer = new DefaultMQProducer("group_test");

        // 声明namesrv地址
        producer.setNamesrvAddr("localhost:9876");

        // 启动实例
        producer.start();

        // 设置消息的topic,tag以及消息体
        Message msg = new Message("topic_test", "tag_test", "消息内容".getBytes(StandardCharsets.UTF_8));

        // 发送消息，并设置10s连接超时
        SendResult send = producer.send(msg, 10000);
        System.out.println("发送结果："+send);

        // 关闭实例
        producer.shutdown();
    }

2、当然，实际生产中，最好不要每次发送消息都创建新的生产者，于是我们需要将其封装成一个bean，实现起来也很简单

@Configuration
public class RocketMQConfig {
    // 配置项大家自定义即可
    @Value("${rocketmq.name-server}")
    private String nameSrvAddr;

    @Value("${rocketmq.producer.group}")
    private String producerGroup;

    private DefaultMQProducer producer;

    @Bean
    public DefaultMQProducer mqProducer(){
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(nameSrvAddr);
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return producer;
    }

}

3、调用：

 @Resource
    private DefaultMQProducer mqProducer;

    @GetMapping("send2")
    public String send2(){
        org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message("topic_test", "tag_test", "消息内容".getBytes(StandardCharsets.UTF_8));
        try {
            mqProducer.send(msg, 1000);
        } catch (Exception e) {
            e.printStackTrace();
            return "fail";
        }
        return "success";
    }

4、因为没有实现DefaultMQProducer的自动关闭，如果要进一步优化，大家可以利用InitializingBean, DisposableBean两个接口实现afterPropertiesSet和destroy，实现资源的自动启动、自动关闭，比如通过destroy实现producer的自动关闭：

    @Override
    public void destroy() {
        if (Objects.nonNull(this.producer)) {
            this.producer.shutdown();
        }
    }

5、发送结果如下所示

SendResult [sendStatus=SEND_OK, msgId=7F00000135DA18B4AAC20E15FE090000, offsetMsgId=C0A8F40100002A9F000000000000C307, messageQueue=MessageQueue [topic=topic_test, brokerName=broker, queueId=15], queueOffset=7]

1.1.1 消息重试设置

实际生产时，可能因为网络波动等不确定原因，导致服务连接出现短时问题，如果恰巧这时我们发送了消息，就会导致发送失败。所以一般我们还会设置消息的重试，只需要调用setRetryTimesWhenSendFailed方法即可

// 设置发送失败时重试次数
producer.setRetryTimesWhenSendFailed(2);

1.2 消息异步发送

消息同步发送可以保证顺序性，但是部分场景我们无需考虑发送次序，并且同步发送如果前面的发送阻塞了，反而会影响后续的发送效率。于是异步发送就产生了。

1、通过查看DefaultMQProducer类可以看到，发送方法中有一个SendCallback参数，这实际就是异步发送的回调方法参数，通过他我们就可以实现异步发送了

2、实现代码

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;


/**
 * @author benjamin_5
 * @Description
 * @date 2024/2/25
 */
public class Producer2AsyncDemo {

    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("group_test");

        producer.setNamesrvAddr("127.0.0.1:9876");

        // 启动实例
        producer.start();

        for (int i = 0; i < 10; i++) {
            Message msg = new Message("topic_test", "tag_test", ("hello "+ i).getBytes(StandardCharsets.UTF_8));
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(new String(msg.getBody(), StandardCharsets.UTF_8)+"发送结果："+sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    System.out.println("异常："+e);
                    e.printStackTrace();
                }
            });

        }
        // 等待发送完再结束主线程
        Thread.sleep(10000);

        producer.shutdown();

    }
}

3、通过观察发送结果，我们也能看出是无序的异步发送

1.3 消息单向发送

单向消息主要适用于非核心日志等允许消息丢失的场景，其方法是没有返回值的，也就是拿不到响应结果，当然好处就是：快！

单向消息支持同步和异步，通过参数可以了解

示例代码如下：

public class Producer5OneWayDemo {

    public static void main(String[] args) throws Exception{

        // 声明group
        DefaultMQProducer producer = new DefaultMQProducer("group_test");

        // 声明namesrv地址
        producer.setNamesrvAddr("localhost:9876");

        // 设置重试次数
        producer.setRetryTimesWhenSendFailed(2);
        // 启动实例
        producer.start();

        // 设置消息的topic,tag以及消息体
        Message msg = new Message("topic_test", "tag_test", "单向消息".getBytes(StandardCharsets.UTF_8));

        // 发送消息
        producer.sendOneway(msg);

        // 关闭实例
        producer.shutdown();
    }
}

1.4 消息顺序发送

官方文档：https://rocketmq.apache.org/zh/docs/4.x/producer/03message2

顺序消息在对流程有严格顺序要求的业务场景有广泛应用，比如订单的创建、付款、发货、签收，这几个状态流程必须保持顺序。

而顺序消息的实现也分成两部分：顺序发送、顺序消费。本章我们主要针对顺序发送进行讲解，顺序消费我们将在下一章继续说明。这里需要注意的是，顺序发送不等于同步发送，虽然同步发送也保持一定的顺序性，但想象如果多个线程发送多个同步消息，那消息还具有顺序性吗？这就不一定了。那顺序发送就是单线程的同步发送吗？

1、单线程的同步发送其实也不能满足顺序性，比如如下的场景，生产者单线程同步发送了两条消息A1,A2, MQ会将其负载到不同的队列存储，然后消费者可能有多个，消费时就容易产生同时从这两个队列消费，从而就不能保证A1一定在A2之前被消费掉。于是单线程的同步发送其实并不能保证顺序性，那么怎么处理呢？

2、既然会发送到不同的队列导致同时消费的可能性，那我们就确保同一数据发送到同一队列，比如同一订单的创建、付款、签收都发送到一个队列中，如此来保证发送的顺序性

3、所以，顺序发送的重点，也在于让同类消息发送到同一队列上，rocketmq中是通过队列选择器MessageQueueSelector来实现的

我们查看send方法，需要提供3个参数：消息、队列选择器和一个arg。前两个参数可以理解，那么这个arg是什么呢，其实就是我们用于区分消息类型的标识，比如为了防止上述的混淆，我们要将同一订单的不同状态的消息都发送到同一个队列中，那么我们就可以以订单号作为这个标识，其目的就是将同一类型的消息通过这个标识进行区分。

4、在发送消息之间，我们需要创建队列，并且将队列指定为顺序队列，即创建队列时指定–order参数为true
（1）我们需要通过namesrv内置的mqadmin工具来实现指定
（2）进入namesrv，在其安装目录的bin目录下执行（如果你和我一样是通过docker安装的rocketmq，那直接进入docker namesrv的容器即可，进入后的当前目录就是bin目录）

./mqadmin updateTopic -c DefaultCluster -t topic_order -o true -n localhost:9876

-n NameServer的地址和端口
-c 指定集群名称
-t 指定主题名称
-w 指定队列的数量，如果要保证全局顺序性，可以设置队列数为1，以此来避免多队列产生的非顺序性问题

5、其次如果需要保证严格的顺序性，还需要在namesrv中配置orderMessageEnable 和 returnOrderTopicConfigToBroker 是 true

（注：默认配置文件为namesrv.properties（通过./mqnamesrv -p即可查看配置文件路径），也可通过创建自定义配置文件namesrv.conf，启动namesrv时指定配置文件nohup sh bin/mqnamesrv -c conf/namesrv.conf &，本文示例无需配置该项也可保证顺序性，但生产时建议配置）

orderMessageEnable=true
returnOrderTopicConfigToBroker=true

6、整体的顺序发送代码如下，这里我简单使用一个orderId作为区分标识（也叫区分键），将奇数和偶数消息分别视为一种消息，大家实际应用时可以根据自己的业务调整。

其MessageQueueSelector对象的定义，主要是实现其select方法，这里就是通过arg参数，做取余运算，进行队列的选择。当然大家也可以用arg的hashcode来作为标识处理

 public static void main(String[] args) throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("group_test");
        // 声明namesrv地址
        producer.setNamesrvAddr("localhost:9876");
        // 设置重试次数
        producer.setRetryTimesWhenSendFailed(2);
        // 启动实例
        producer.start();

        // 设置消息的topic,tag以及消息体
        Message msg = new Message("topic_test", "tag_test", "消息内容".getBytes(StandardCharsets.UTF_8));

        // 要求发送顺序：i为偶数先发，然后按照由小到大顺序发送
        for (int i = 0; i < 10; i++) {
            // 模拟偶数、奇数分别属于一类消息
            int orderId = i % 2;

            SendResult result = producer.send(msg, new MessageQueueSelector() {
                /**
                 *
                 * @param list 消息队列集合
                 * @param message 消息
                 * @param arg send方法中传入的第三参数，即orderId参数，orderId可以是Object类型
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object arg) {
                    Integer orderId = (Integer) arg;
                    int index = orderId % list.size();
                    return list.get(index);
                }
            }, orderId);
            System.out.println("发送结果："+result.toString());
        }
        producer.shutdown();
    }

7、提前用顺序消费的代码来消费验证下

public class Consumer2OrderDemo {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_test");

        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 集群消费模式
        consumer.setMessageModel(MessageModel.CLUSTERING);

        // 设置topic
        consumer.subscribe("topic_order", "*");

        // 注册回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
              @Override
              public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                  byte[] body = list.get(0).getBody();
                  System.out.println("接收消息："+new String(body, StandardCharsets.UTF_8));
                  return ConsumeOrderlyStatus.SUCCESS;
              }
        });

        // 启动消费者实例
        consumer.start();

        Thread.sleep(10000);
    }
}

消费到的消息如下所示，可以看到奇偶是分别保持顺序的，即：0,2,4,6,8 和1,3,5,7,9

最后总结一下，顺序发送的前提要满足两点：

单一生产者：不同的生产者部署在不同的服务器，哪怕使用相同的区分键，也可能导致不同生产者之间的消息无法区分顺序性
串行发送：多线程发送可能会导致并发发送，从而导致破坏消息顺序性

因此rocketmq的顺序消息，实际是基于队列的顺序消息，不同队列之间的消息是不满足顺序性的。这一点大家要注意。

1.5 消息批量发送

官方文档：https://rocketmq.apache.org/zh/docs/4.x/producer/05message4

批量消息相对比较简单，从发送方法中可以看到支持集合形式的message对象

因此我们只需要传入List即可

public class Producer4BatchDemo {

    public static void main(String[] args) throws Exception {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("group_test");
            // 声明namesrv地址
            producer.setNamesrvAddr("localhost:9876");
            // 设置重试次数
            producer.setRetryTimesWhenSendFailed(2);
            producer.start();

            String topic = "topic_test";
            List<Message> messages = new ArrayList<>();
            messages.add(new Message(topic, "Tag",  "消息1".getBytes()));
            messages.add(new Message(topic, "Tag",  "消息2".getBytes()));
            messages.add(new Message(topic, "Tag",  "消息3".getBytes()));
            SendResult send = producer.send(messages);

            System.out.println("发送结果："+send);
            producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

需要注意的是:
（1）批量消息的大小不能超过 1MiB（否则需要自行分割）
（2）同一批 batch 中 topic 必须相同
（3）批量消息是不支持延迟消息的

1.6 消息延迟发送

官方文档：https://rocketmq.apache.org/zh/docs/4.x/producer/04message3

rocketMQ的延迟消息时间不能自定义，但是可以在预设的16个等级中选择，也能够满足我们大部分的业务场景

延迟消息的发送，只需要在message中指定延迟时间等级即可，通过setDelayTimeLevel方法实现，示例代码如下

public class Producer6ScheduledDemo {

    public static void main(String[] args) {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("group_test");
            // 声明namesrv地址
            producer.setNamesrvAddr("localhost:9876");
            // 设置重试次数
            producer.setRetryTimesWhenSendFailed(2);
            producer.start();

            int totalMessagesToSend = 100;
            for (int i = 0; i < totalMessagesToSend; i++) {
                Message message = new Message("topic_test", ("延迟消息 " + i).getBytes());
                
                message.setDelayTimeLevel(3);
                // Send the message
                SendResult send = producer.send(message);
                System.out.println("发送结果："+send);
            }
            
            producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

1.7 事务消息发送

rocketmq的一大特性就是支持事务消息，他的原理是利用half message来实现的，这个具体原理我们后续单独讲解，今天先关注他的使用

事务消息的发送不再使用 DefaultMQProducer，而是使用 TransactionMQProducer 进行发送，我们还可以设置事务回查的线程池，当然如果不设置也会默认生成一个

1、首先需要实现 TransactionListener 接口，并传入 TransactionMQProducer

事务消息和数据库事务类似，也是通过两阶段提交的形式来实现的，所以其实现方法中要重写executeLocalTransaction和 checkLocalTransaction方法

executeLocalTransaction方法是用于执行消费消息以及后续的其他事务事件，然后返回一个事物状态，用于告诉rocketmq这条事务消息是可以正常提交了还是需要回滚。

2、在讲具体实现代码前我们先理清楚几个概念，帮助大家理解为什么要书写这些代码，首先查看状态枚举LocalTransactionState可以知道，executeLocalTransaction方法返回值一共有3种状态：

COMMIT_MESSAGE

提交状态，事务正常进行，一般是本地事务执行成功后进行设置。告知broker提交该事务消息，然后消费者可以消费该消息，当然此时消费者已经执行完本地事务了，再消费可以根据业务逻辑进行后续的逻辑处理，如果没有相关逻辑了忽略消息即可

ROLLBACK_MESSAGE

回滚状态，事务撤回，broker将删除当前half消息，一般是本地事务执行失败后进行设置

UNKNOW

未知状态，固定时间后Broker端会通过checkLocalTransaction方法进行消息回查，根据回查结果来判断该消息是提交还是回滚

3、要理解checkLocalTransaction方法，我们还得先理解什么是消息回查？

有一种场景：当本地事务执行完成后，在返回COMMIT_MESSAGE或ROLLBACK_MESSAGE状态时，因为网络波动或者broker服务挂了，导致broker没有正常收到这个状态，从而没有及时把half message进行提交或回滚，这时就需要有个定时巡查机制，来检查这些没有正常收到提交状态的消息的实际状态到底是什么，这个巡查机制就是消息回查。因此checkLocalTransaction方法中就要书写检查本地事务状态的方法，比如事务是对订单提交消息的消费，那么就去查询订单状态，如果已经是提交状态那么就返回COMMIT_MESSAGE，否则就返回ROLLBACK_MESSAGE

最后总结一下，executeLocalTransaction方法用于书写消费消息，执行本地事务的代码，并最终返回一个状态，本地事务执行成功则返回COMMIT_MESSAGE状态，执行失败则返回ROLLBACK_MESSAGE，UNKNOW状态实际书写时是基本用不到的。checkLocalTransaction状态用于消息状态回查，需要在该方法中提供一个查询本地事务执行状态的方法，然后返回实际执行状态，本地事务执行成功返回COMMIT，执行失败返回ROLLBACK

4、实现代码，申明TransactionListener接口，实现executeLocalTransaction和checkLocalTransaction方法，生产中可以将TransactionListenerImpl申明为bean, 然后引入其他dao类或者service类处理本地事务

class TransactionListenerImpl implements TransactionListener {

        @Override
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            // TODO 执行本地事务（书写你自己的本地事务逻辑）
            try{
                String body = new String(msg.getBody());
                int i = Integer.parseInt(body);
                // 模拟偶数执行成功，奇数执行失败
                if(i % 2 == 0){
                    System.out.println("本地事务执行成功："+body);
                    // 执行成功
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else{
                    System.out.println("本地事务执行失败："+body);
                    // 执行失败
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

            }catch (Exception e){
                e.printStackTrace();
                // 执行失败
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }

        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
            // TODO 去缓存或者数据库查询当前消息的实际状态

            // 模拟查询到状态为1
            Integer status = 1;

            // 不同实际状态对应的消息状态
            if (null != status) {
                switch (status) {
                    case 1:
                        return LocalTransactionState.COMMIT_MESSAGE;
                    case 2:
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    default:
                        return LocalTransactionState.COMMIT_MESSAGE;
                }
            }
            return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

6、消息发送方法，事务消息不能再用DefaultMQProducer了，得用TransactionMQProducer

这里额外申明了回查线程池executorService，当然如果不申明rocektmq也会默认创建一个的

public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
        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("transaction-msg-check-thread");
                return thread;
            }
        });

        producer.setExecutorService(executorService);
        producer.setTransactionListener(transactionListener);
        producer.setNamesrvAddr("localhost:9876");

        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {
                Message msg =
                        new Message("topic_test", tags[i % tags.length], "KEY" + i,
                                (""+i).getBytes());
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);

                Thread.sleep(10);
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        }

        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();
    }

7、执行消息发送结果

8、消费消息，可以看到只有偶数消息正常提交消费了，奇数都被回滚了

3. 总结

至此我们针对java client实现各类消息发送的方法就梳理完成了，但实际工作中，我们现在更加常用的是基于springboot框架，而rocektmq也有专门针对springboot框架进行集成，实现起来更加简单，下一期我们重点讲解springboot集成实现消息发送

本文演示源码见：https://gitee.com/wuhanxue/wu_study/tree/master/demo/rocketmq_demo

文章来源:https://blog.csdn.net/qq_24950043/article/details/138338692
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：jacktools123@163.com进行投诉反馈，一经查实，立即删除！

标签：

上一篇：JavaScript在数据可视化领域的探索与实践
下一篇：华为OD机试 - 分解连续正整数组合 - 数学推导（Java 2024 E卷 100分）

首页 > 基础资料博客日记

RocketMQ快速入门：集成java客户端实现各类消息发送｜异步、同步、顺序、单向、延迟、事务（五）附带源码

目录

0. 引言

1. 集成rocketMQ

1.1 消息同步发送

1.1.1 实现步骤

1.1.1 消息重试设置

1.2 消息异步发送

1.3 消息单向发送

1.4 消息顺序发送

1.5 消息批量发送

1.6 消息延迟发送

1.7 事务消息发送

3. 总结

相关文章

最新发布

点击排行

本站推荐

标签云

首页 > 基础资料 博客日记

RocketMQ快速入门：集成java客户端实现各类消息发送｜异步、同步、顺序、单向、延迟、事务（五）附带源码

目录

0. 引言

1. 集成rocketMQ

1.1 消息同步发送

1.1.1 实现步骤

1.1.1 消息重试设置

1.2 消息异步发送

1.3 消息单向发送

1.4 消息顺序发送

1.5 消息批量发送

1.6 消息延迟发送

1.7 事务消息发送

3. 总结

相关文章

最新发布

点击排行

本站推荐

标签云

首页 > 基础资料博客日记