RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

>>强大,10k+点赞的 SpringBoot 后台管理系统竟然出了详细教程!

RocketMQ系列第二篇。熬夜不易,且行且珍惜。

通过RocketMQ的API来直观的感受一下,RocketMQ是怎样的进行消息生产和消费的。首先安装一个RocketMQ的扩展rocketmq-console控制台,然后通过API演示RocketMQ的消息发送模式和消费消息模式,最后介绍一下消费者如何通过TAG、SQL表达式来过滤消息。

  • rocketmq-console
  • 发送消息的方式
  • 消费消息
  • TAG过滤
  • SQL表达式过滤

0x01 安装RocketMQ扩展-rocketmq-console

RocketMQ官方GitHub上有一个项目rocketmq-externals,提供了很多扩展:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息
RocketMQ扩展

其中,rocketmq-console能够为我们直观的展示RocketMQ集群分部情况、Producer、Consumer、Topic等等,下面我们来装一个看看长什么样。

为了方便,这次我用Docker进行安装,到Docker Hub上找到rocketmq-console的Docker官方镜像:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

使用Docker安装RocketMQ控制台

# 拉取镜像
docker pull apacherocketmq/rocketmq-console:2.0.0

# 启动
docker run -e "JAVA_OPTS=-Drocketmq.namesrv.addr=192.168.2.110:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -p 8080:8080 -t apacherocketmq/rocketmq-console:2.0.0

启动成功,出现熟悉的打印信息:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

通过http://192.168.2.110:8080访问控制台:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

这个控制台做的还是挺炫酷的!

Docker系列大纲已就绪,后面会输出Docker系列文章,欢迎关注并督促我,^ _ ^

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

0x02 Producer发送消息

2.1 引入jar包

首先需要引入RocketMQ Client的jar包,这个注意一下版本就行了,最好和安装的RocketMQ版本一致,所以这里选择4.7.1版本:

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

2.2 同步消息API

RocketMQ是处理各种消息的,消息来自于Producer,那么要发送消息,自然就能想到要有发送消息的生产者实例,API提供了DefaultMQProducer这个类,其构造方法如下:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

我们先new一个Producer实例出来,先能发送消息再说。

public class SyncMsgProducer {
    public static void main(String[] args) throws Exception {
        //实例化消息生产者,参数是producerGroup
        DefaultMQProducer producer = new DefaultMQProducer("laogong");
        //设置nameserver的地址
        producer.setNamesrvAddr("192.168.2.110:9876");
        //启动producer
        producer.start();

        //发送消息
//        Message msg = new Message("xiaoxianrou", "这是我的第一次".getBytes());
//        SendResult sendResult = producer.send(msg);
//        System.out.printf("%s%n", sendResult);
        //批量发送
        List<Message> msgs = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            Message msg = new Message("xiaoxianrou", ("这是我的第" + i + "次").getBytes());
            msgs.add(msg);
        }
        SendResult sendResult = producer.send(msgs);
        System.out.printf("%s%n", sendResult);

        //关闭producer
        producer.shutdown();
        System.out.println("已关闭producer实例");
    }
}

这段代码提供了一次发送一条消息和批量发送消息的示例,运行它:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

可以看到,send(msg)方法「同步发送消息」,有一个返回值,也就是说消息发送中一定会给客户端一个状态,等broker说我收到了之后,返回一个SendResult,在此后这条消息就和Producer没关系了。

同步发送过程中Producer进入「同步等待状态」「可以保证消息投递一定到达」

这种「可靠性」同步地发送方式使用的比较广泛,比如:重要的消息通知,短信通知。

可以在控制台看一下发送的message:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

进一步查看消息的详细信息:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

最下面的TraceList展示了消息的消费情况,由于我们还没有消费它,所以这里没有记录。

PS:出现

org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout

的解决方案

  1. 修改/usr/local/rocketmq/conf/broker.conf,添加brokerIP1=192.168.2.110,IP地址是自己的虚拟机IP地址
  2. 重启nameserver
  3. 重启broker:./mqbroker -n 192.168.2.110:9876 -c /usr/local/rocketmq/conf/broker.conf

2.3 批量消息发送

上面的例子中提到了send方法可以批量发送消息,当一次性发送很多条消息时,可以多条消息打包一起发送,「减少网络传输次数提高效率」

producer.send(Collection c) 方法可以接受一个集合,实现批量发送:

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

批量发送需注意:

  • 批量消息要求必要具有同一topic、相同消息配置
  • 不支持延时消息
  • 这一批消息的总大小不应超过4MB
  • 如果不确定是否超过限制,可以手动计算大小分批发送

2.4 异步消息API

Producer的API中send方法也提供了异步的发送方式:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

「show you the code」

int messageCount = 100;
// 根据消息数量实例化倒计时计算器
final CountDownLatch2 countDownLatch = new CountDownLatch2(messageCount);
for (int i = 0; i < messageCount; i++) {
    final int index = i;
    // 创建消息,并指定Topic,Tag和消息体
    Message msg = new Message("xiaoxianrou",
            "TagA",
            "OrderID188",
            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
    // SendCallback接收异步返回结果的回调
    producer.send(msg, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
        }
        @Override
        public void onException(Throwable e) {
            System.out.printf("%-10d Exception %s %n", index, e);
            e.printStackTrace();
        }
    });
}
// 等待5s
countDownLatch.await(5, TimeUnit.SECONDS);
// 如果不再发送消息,关闭Producer实例。
producer.shutdown();

运行结果:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

控制台查看消息详情:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

以上消息是通过异步的方式生成的,异步消息通常用在「对响应时间敏感」的业务场景,即「发送端不能容忍长时间地等待Broker的响应」

「想要快速发送消息,又不想丢失消息的时候可以使用异步消息。」

2.5 单向消息API

只发送消息,不等待服务器响应,只发送请求不等待应答。

此方式发送消息的过程耗时非常短,一般在微秒级别。

其API就是调用sendOneway方法:

for (int i = 0; i < 100; i++) {
    // 创建消息,并指定Topic,Tag和消息体
    Message msg = new Message("xiaoxianrou" ,
            "TagA",
            ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
    );
    // 发送单向消息,没有任何返回结果
    producer.sendOneway(msg);
}

0x03 Consumer消费消息

3.1 消息消费模式

消息消费模式由消费者Consumer来决定,可以由消费者设置MessageModel来决定消息模式。

消息模式默认为集群消费模式,此外还有广播消费模式。

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

3.1.1 集群消费模式

集群消费消息是指「集群化部署消费者」

当使用集群消费模式时,MQ认为「任意一条消息只需要被集群内的任意一个消费者处理即可」

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息
集群消费模式

集群消费模式的特点:

  • 每条消息只需要被处理一次,broker只会把消息发送给消费集群中的一个消费者
  • 在消息重投时,不能保证路由到同一台机器上
  • 消费状态由broker维护

消费者消费消息代码:

public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("laogong-consumer");
        consumer.setNamesrvAddr("192.168.2.110:9876");
        //订阅topic,根据tag过滤消息
        consumer.subscribe("xiaoxianrou""*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), list);
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //设置消费模式,默认就是CLUSTERING
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 启动Consumer实例
        consumer.start();
        System.out.println("consumer started.");
    }
}

默认就是集群消费模式,运行结果:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

看一下控制台:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

可以看出,之前Producer产生的消息状态已变成「consumed」了。

3.1.2 广播消费模式

当使用广播消费模式时,MQ会「将每条消息推送给集群内所有注册过的客户端,保证消息至少被每台机器消费一次」

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息
广播消费模式

API设置广播消费模式很简单:

consumer.setMessageModel(MessageModel.BROADCASTING);

广播消费模式的特点:

  • 消费进度由consumer维护
  • 保证每个消费者消费一次消息
  • 消费失败的消息不会重投

0x04 关于TAG

前面的案例提到了tag,Consumer在订阅的时候,除了订阅topic外,还可以指定tag,对消息进行过滤。

比如,Producer发送topic为xiaoxianrou,tag为TagATagB的消息,Consumer只订阅TagA,那么这个Consumer则只处理TagA的消息。

我们还是通过API和控制台来看一下消息状态。

生产者产生的消息:

// 创建消息,并指定Topic,Tag和消息体
Message msg = new Message("xiaoxianrou",
        "TagA",
        "OrderID188",
        ("laogong" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
// 设置属性
msg.putUserProperty("money", String.valueOf(i));

// TagB
Message msg = new Message("xiaoxianrou",
        "TagB",
        "OrderID288",
        ("laogong" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
msg.putUserProperty("money", String.valueOf(i));

消费者消费,指定TagA:

//订阅topic,根据tag过滤消息
consumer.subscribe("xiaoxianrou""TagA");

来看一下控制台:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

由于Consumer订阅topic的时候,指定了TagA,所以猜测TagB应该会被过滤掉,我们来验证一下,先看一条TagA的消息消费情况:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

TagA的消息均是CONSUMED,已消费状态,再来看一条TagB的消息:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

被过滤了。

0x05 SQL表达式过滤消息

消费者收到包含TagA或TagB的消息,但限制是一条消息只能有一个标签,而这对于复杂的情况可能无效。

在这种情况下,可以使用SQL表达式筛选出消息。

首先需要配置一下/usr/local/rocketmq/conf/broker.conf,添加:

enablePropertyFilter=true

然后指定broker.conf,重启broker:

./mqbroker -n 192.168.2.110:9876 -c /usr/local/rocketmq/conf/broker.conf

重启之后,控制台集群会显示该属性:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

前文所述案例中我设置了:

msg.putUserProperty("money", String.valueOf(i));

其中TagA的money是0~49,已经被consumer消费了,现在我再开一个通过sql表达式过滤出money大于49的消息,API如下:

//订阅topic,根据sql表达式过滤消息
MessageSelector selector = MessageSelector.bySql("money > 49");
consumer.subscribe("xiaoxianrou", selector);

消费完了再来看,TagB的状态:

RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息

变成已消费了。


首发公众号 「行百里er」 ,欢迎老铁们关注阅读指正。代码仓库 「GitHub」 github.com/xblzer/JavaJourney


原文始发于微信公众号(行百里er):RocketMQ入坑系列(二)近距离感受RocketMQ如何收发消息