系列十三、Java操作RocketMQ之带Key的消息

一、概述

        RocketMQ中的消息，默认会有一个messageId当做消息的唯一标识，我们也可以给消息携带一个key，用作唯一标识或者业务标识，包括在控制面板（Dashboard，RocketMQ的一个可视化面板）中也可以使用messageId或者key来进行查询。

二、案例代码

2.1、pom

        同案例五

2.2、RocketMQConstant

        同案例五

2.3、KeyConsumer

package org.star.key.consumer;import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.star.constants.RocketMQConstant;import java.util.List;/*** @Author: 一叶浮萍归大海* @Date: 2023/9/8 10:15* @Description: 带有Key的消息消费者*/
@Slf4j
public class KeyConsumer {public static void main(String[] args) throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("KeyConsumerGroup");consumer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);consumer.subscribe("KeyTopic","*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {log.info("消费者[KeyConsumer]正在消费消息，当前线程：{}，消息内容：{}，标签：{}，key：{}",Thread.currentThread().getName(), StrUtil.utf8Str(list.get(0).getBody()),list.get(0).getTags(),list.get(0).getKeys());return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();log.info("KeyConsumer start success");}}

2.4、KeyProducer

package org.star.key.producer;import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.star.constants.RocketMQConstant;import java.nio.charset.StandardCharsets;
import java.util.UUID;/*** @Author: 一叶浮萍归大海* @Date: 2023/9/8 10:06* @Description: 带有Key的消息生产者*/
@Slf4j
public class KeyProducer {public static void main(String[] args) throws Exception {DefaultMQProducer producer = new DefaultMQProducer("KeyProducerGroup");producer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);producer.start();log.info("KeyProducer start success!");String key = UUID.randomUUID().toString().replaceAll("-","");Message message = new Message("KeyTopic","KeyTag",key,"我是一个带有标记和key的消息".getBytes(StandardCharsets.UTF_8));SendResult sendResult = producer.send(message);log.info("发送结果：{}，消息ID：{}，队列ID：{}",sendResult.getSendStatus(),sendResult.getMsgId(),sendResult.getMessageQueue().getQueueId());producer.shutdown();}}

2.5、控制台打印

# 生产者端
10:23:01.171 [main] INFO org.star.key.producer.KeyProducer - KeyProducer start success!
10:23:01.657 [main] INFO org.star.key.producer.KeyProducer - 发送结果：SEND_OK，消息ID：0AA86761652418B4AAC22646EA120000，队列ID：2# 消费者端
10:23:33.015 [ConsumeMessageThread_1] INFO org.star.key.consumer.KeyConsumer - 消费者[KeyConsumer]正在消费消息，当前线程：ConsumeMessageThread_1，消息内容：我是一个带有标记和key的消息，标签：KeyTag，key：bbf5efa94d0e473987f5718f3c023c9c