[Java知识库]10. RocketMQ其他功能和常见问题

RocketMQ 其他功能

消息轨迹

暂略

配置

Broker 端服务器开启配置：traceTopicEnable=true

traceTopicEnable=true

? RocketMQ 集群中每一个 Broker 节点均用于存储 Client 端收集并发送过来的消息轨迹数据。因此，对于 RocketMQ 集群中的 Broker 节点数量并无要求和限制。

? 对于消息轨迹数据量较大的场景，可以在 RocketMQ 集群中选择其中一个 Broker 节点专用于存储消息轨迹，使得用户普通的消息数据与消息轨迹数据的物理 IO 完全隔离，互不影响。在该模式下，RockeMQ 集群中至少有两个 Broker 节点，其中一个 Broker 节点定义为存储消息轨迹数据的服务端。

保存消息轨迹的 Topic 定义

? RocketMQ 的消息轨迹特性支持两种存储轨迹数据的方式：

? 系统级的 TraceTopic

? 在默认情况下，消息轨迹数据是存储于系统级的 TraceTopic 中(其名称为：RMQ_SYS_TRACE_TOPIC**，**队列个数为 1)。该 Topic 在 Broker 节点启动时，会自动创建出来（如上所叙，需要在 Broker 端的配置文件中将 traceTopicEnable 的开关变量设置为 true）。

? 用户自定义的 TraceTopic

? 如果用户不准备将消息轨迹的数据存储于系统级的默认 TraceTopic，也可以自己定义并创建用户级的 Topic 来保存轨迹（即为创建普通的 Topic

用于保存消息轨迹数据）。自定义的话需要在 Client 客户端的处理的时候自定义的 TraceTopic。具体见案例。一般推荐使用系统及的 TraceTopic。

案例

? RocketMQ 在消息审核消费时采用对原来接口增加一个开关参数(enableMsgTrace)来实现消息轨迹是否开启；并新增一个自定义参数

(customizedTraceTopic)来实现用户存储消息轨迹数据至自己创建的用户级 Topic。

public class TraceProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("produce", true, "TopicTrace_111"); // 开关, 默认是false
        //DefaultMQProducer producer = new DefaultMQProducer("produce",true,"TopicTrace_111");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        for (int i = 0; i < 3; i++) {
            try {
                Message msg = new Message("TopicTrace",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        producer.shutdown();
    }
}

public class TracePushConsumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        // Here,we use the default message track acl topic name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1", true);
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTrace", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

对于定义轨迹的主题，需要先创建这个主题才能收到消息，RocketMQ 不会自动创建该主题。

关键属性

源码解读

代码实现的核心类是 AsyncTraceDispatcher

// org.apache.rocketmq.client.producer.DefaultMQProducer#DefaultMQProducer
public DefaultMQProducer(final String namespace, final String producerGroup, RPCHook rpcHook,
                         boolean enableMsgTrace, final String customizedTraceTopic) {
    this.namespace = namespace;
    this.producerGroup = producerGroup;
    defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
    //if client open the message acl feature
    if (enableMsgTrace) {
        try { // here
            AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(
                producerGroup, TraceDispatcher.Type.PRODUCE, customizedTraceTopic, rpcHook);
            dispatcher.setHostProducer(this.getDefaultMQProducerImpl());
            traceDispatcher = dispatcher;
            //todo 为消息轨迹注册hook,在消息发送前后执行
            this.getDefaultMQProducerImpl().registerSendMessageHook(
                new SendMessageTraceHookImpl(traceDispatcher));
        } catch (Throwable e) {
            log.error("system mqtrace hook init failed ,maybe can't send msg acl data");
        }
    }
}
// org.apache.rocketmq.client.consumer.DefaultMQPushConsumer#DefaultMQPushConsumer
public DefaultMQPushConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook,
                             AllocateMessageQueueStrategy allocateMessageQueueStrategy, boolean enableMsgTrace, final String customizedTraceTopic) {
    this.consumerGroup = consumerGroup;
    this.namespace = namespace;
    this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
    defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
    if (enableMsgTrace) {
        try { // here
            AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(consumerGroup, TraceDispatcher.Type.CONSUME,
                                                                       customizedTraceTopic, rpcHook);
            dispatcher.setHostConsumer(this.getDefaultMQPushConsumerImpl());
            traceDispatcher = dispatcher;
            this.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(
                new ConsumeMessageTraceHookImpl(traceDispatcher));
        } catch (Throwable e) {
            log.error("system mqtrace hook init failed ,maybe can't send msg acl data");
        }
    }
}

SendMessageHook

public interface SendMessageHook {
    String hookName();

    void sendMessageBefore(final SendMessageContext context);

    void sendMessageAfter(final SendMessageContext context);
}

ConsumeMessageHook

public interface ConsumeMessageHook {
    String hookName();

    void consumeMessageBefore(final ConsumeMessageContext context);

    void consumeMessageAfter(final ConsumeMessageContext context);
}

? 通过实行上述两个接口，可以实现在消息发送、消息消费前后记录消息轨迹，为了不明显增加消息发送与消息消费的时延，记录消息轨迹最好使用异步发送模式。

核心步骤如下：

1：遍历收集的消息轨迹数据

2：获取存储消息轨迹的 Topic

3：对 TraceContext 进行编码，这里是消息轨迹的传输数据。

4：将编码后的数据发送到 Broker 服务器。

public void start(String nameSrvAddr, AccessChannel accessChannel) throws MQClientException {
    if (isStarted.compareAndSet(false, true)) {
        traceProducer.setNamesrvAddr(nameSrvAddr);
        traceProducer.setInstanceName(TRACE_INSTANCE_NAME + "_" + nameSrvAddr);
        traceProducer.start();
    }
    this.accessChannel = accessChannel;
    this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId);
    this.worker.setDaemon(true);
    this.worker.start();
    this.registerShutDownHook();
}

权限控制

? 在 RocketMQ4.4.0 版本升级中加入了 ACL 权限管控，ACL 是 access control list 的简称，俗称访问控制列表。访问控制，基本上会涉及到用户、资源、权限、角色等概念。

? 用户的概念，即支持用户名、密码。

? 资源：需要保护的对象，消息发送涉及的 Topic、消息消费涉及的消费组，应该进行保护，故可以抽象成资源。

? 权限：针对资源，能进行的操作。 角色：RocketMQ 中，只定义两种角色：是否是管理员。

配置

? acl 默认的配置文件名：plain_acl.yml,需要放在${ROCKETMQ_HOME}/store/config 目录下需要使用 acl 必须在服务端开启此功能，在 Broker 的配置文件中配置，aclEnable = true 开启此功能

? RocketMQ 的权限控制存储的默认实现是基于 yml 配置文件。用户可以动态修改权限控制定义的属性，而不需重新启动 Broker 服务节点，因为 Broker 端有文件监听机制，每隔 500ms 监听、处理、加载文件的变更内容。

? 如果 ACL 与高可用部署(Master/Slave 架构)同时启用，那么需要在 Broker Master 节点的${ROCKETMQ_HOME}/store/conf/plain_acl.yml 配置文件中设置

? 全局白名单信息，即为将 Slave 节点的 ip 地址设置至 Master 节点 plain_acl.yml 配置文件的全局白名单中。

? plain_acl.yml 文件中相关的参数含义及使用

DENY 拒绝

ANY PUB 或者 SUB 权限

PUB 发送权限

SUB 订阅权限

案例

在 Broker 的配置文件中配置，aclEnable = true 开启此功能

代码得加入一个返回 RPCHook 的方法

new DefaultMQProducer("benchmark_transaction_producer", config.aclEnable ? AclClient.getAclRPCHook() : null);

public class AclClient {

    private static final String ACL_ACCESS_KEY = "rocketmq2";

    private static final String ACL_SECRET_KEY = "12345678";

    static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(ACL_ACCESS_KEY, ACL_SECRET_KEY));
    }
}

常见问题

MQ 百万消息积压处理

? 发生了线上故障，几千万条数据在 MQ 里积压很久。是修复 consumer 的问题，让他恢复消费速度，然后等待几个小时消费完毕？这是个解决方案。 不过有时候我们还会进行临时紧急扩容。

? 一个消费者一秒是 1000 条，一秒 3 个消费者是 3000 条，一分钟是 18 万条。1000 多万条，所以如果积压了几百万到上千万的数据，即使消费者恢复 了，也需要大概 1 小时的时间才能恢复过来。

? 一般如果消息不重要的话就在 consume 上直接释放掉。 如果 topic 的 messageQueue 设置的比较多，比如设置了 20 个，consume 实例只有 4 个，那么每个 consume 实例对应 5 个 messageQueue，这个时候 可以申请临时加机器，增加 consume 实例为 20 个，达到快速消费的目的。

? 如果 messageQueue 设置的比较少，比如只设置了 4 个，那么这个时候就不能通过加 consume 机器来解决了，这时候就需要修改消费者代码了，不再 消费者消费，而是把要消费的消息放到 mq 的另一个 topic 中，这个 topic 设置 20 个 messageQueue，对应 20 个 consume 实例，进行消费

消费幂等

消息重复的场景

生产者

? 重试机制导致的问题，消息成功发送到 MQ 中**，但 MQ 因网络原因未能成功返回，导致重试机制重试机制重复发送到 MQ**

这个我在kafka中遇到过,服务端使用的是kafka-server的2.4.1,但是我们的用户使用的客户端是kafka-client的1.1.0版本,导致客户端不停重试,出现不同数量的重复消息

消费端

? 手动提交 offset 未完成

? 消费者成功消费完消息，未返回 consume_commit 时，系统重启|系统宕机，

? MQ 重新发送消息到同消息组其他消费者机器，导致消息重复

什么是幂等性？

? 对于消息接收端的情况,幂等的含义是采用同样的输入多次调用处理函数,得到同样的结果。例如，一个 SQL 操作

? update stat_table set count= 10 where id =1

? 这个操作多次执行,id 等于 1 的记录中的 count 字段的值都为 10,这个操作就是幂等的,我们不用担心这个操作被重复。

? 再来看另外一个 SQL 操作

? update stat_table set count= count +1 where id= 1;

? 这样的 SQL 操作就不是幂等的,一旦重复,结果就会产生变化。

MVCC

? 多版本并发控制，乐观锁的一种实现，在生产者发送消息时进行数据更新时需要带上数据的版本号，消费者去更新时需要去比较持有数据的版本号，版本号不一致的操作无法成功。例如博客点赞次数自动+1 的接口：

public boolean addCount(Long id, Long version);

update blogTable set count= count+1,version=version+1 where id=321 and version=123

每一个 version 只有一次执行成功的机会，一旦失败了生产者必须重新获取数据的最新版本号再次发起更新。

去重表

? 利用数据库表单的特性来实现幂等，常用的一个思路是在表上构建唯一性索引(也可以使用redis缓存key)，保证某一类数据一旦执行完毕，后续同样的请求不再重复处理了（利用一张日志表来记录已经处理成功的消息的 ID，如果新到的消息 ID 已经在日志表中，那么就不再处理这条消息。）

? 利用 RocketMQ 的 key 值

? 以电商平台为例子，电商平台上的订单 id 就是最适合的 token。当用户下单时，会经历多个环节，比如生成订单，减库存，减优惠券等等。每一个环节执行时都先检测一下该订单 id 是否已经执行过这一步骤，对未执行的请求，执行操作并缓存结果，而对已经执行过的 id，则直接返回之前的执行结果，不做任何操作。这样可以在最大程度上避免操作的重复执行问题，缓存起来的执行结果也能用于事务的控制等。