【消息中间件】默认RocketMQ消息发送者是如何启动的？-阿里云开发者社区

【消息中间件】默认RocketMQ消息发送者是如何启动的？

2022-11-28 237

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

任务调度 XXL-JOB 版免费试用，400 元额度，开发版规格

性能测试 PTS，5000VUM额度

Serverless 应用引擎免费试用套餐包，4320000 CU，有效期3个月

简介： 上一篇文章，主要介绍了RocketMQ消息发送-请求与响应，了解了消息发送的请求参数和响应结果，今天我们主要来学习默认消息发送者的源码，看看消息发送主要是做了那哪些事情。

前言

上一篇文章，主要介绍了RocketMQ消息发送-请求与响应，了解了消息发送的请求参数和响应结果，今天我们主要来学习默认消息发送者的源码，看看消息发送主要是做了那哪些事情。

思考问题

对于过期的请求，发送者是如何处理的？
MQClientFactory的作用是什么？

默认消息发送者-工作原理

源码入口：org.apache.rocketmq.client.producer.DefaultMQProducer#start

DefaultMQProducer producer = new DefaultMQProducer("group1");
producer.setNamesrvAddr("xxx:9876");
producer.start();

serviceState状态为CREATE_JUST，服务将创建未启动默认的状态为CREATE_JUST

private ServiceState serviceState = ServiceState.CREATE_JUST;

检查配置

入口：org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#checkConfig

主要就是为了检查生产者分组是否符合要求

分组不能为空
符合正则规则
字符串长度不能大于255
不能使用默认分组DEFAULT_PRODUCER

private void checkConfig() throws MQClientException {
    Validators.checkGroup(this.defaultMQProducer.getProducerGroup());
    if (null == this.defaultMQProducer.getProducerGroup()) {
        throw new MQClientException("producerGroup is null", null);
    }
    if (this.defaultMQProducer.getProducerGroup().equals(MixAll.DEFAULT_PRODUCER_GROUP)) {
        throw new MQClientException("producerGroup can not equal " + MixAll.DEFAULT_PRODUCER_GROUP + ", please specify another one.",
            null);
    }
}

如果分组不是CLIENT_INNER_PRODUCER，则修改生产者InstanceNama为进程ID

入口：org.apache.rocketmq.client.ClientConfig#changeInstanceNameToPID

InstanceNama：客户端实例名称，是客户端标识 CID 的组成部分

public void changeInstanceNameToPID() {
    if (this.instanceName.equals("DEFAULT")) {
        this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
    }
}

生成MQ客户端工厂MQClientFactory

入口： org.apache.rocketmq.client.impl.MQClientManager#getOrCreateMQClientInstance(org.apache.rocketmq.client.ClientConfig, org.apache.rocketmq.remoting.RPCHook)

创建MQClientInstance实例，才用单例模式，保证JVM中只存在一个实例

将实例保存进ConcurrentMap

private ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable =
    new ConcurrentHashMap<String, MQClientInstance>();

一个clientId只会对应一个实例,根据客户端ID获取实例，如果未获取到，则初始化一个客户端实例

在这里进行了两次instance的检查，利用恶汉模式，确保只存在一个实例

public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
    String clientId = clientConfig.buildMQClientId();
    MQClientInstance instance = this.factoryTable.get(clientId);
    if (null == instance) {
        instance =
            new MQClientInstance(clientConfig.cloneClientConfig(),
                this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
        MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
        //第二次检查,如果出现重复Key，则添加失败
        if (prev != null) {
            instance = prev;
            log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
        } else {
            log.info("Created new MQClientInstance for clientId:[{}]", clientId);
        }
    }
    return instance;
}

向MQClientInstance注册发送者，方便后续网络请求

入口：org.apache.rocketmq.client.impl.factory.MQClientInstance#registerProducer

将生产者保存进ConcurrentMap

private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = 
                            new ConcurrentHashMap<String, MQProducerInner>();

public synchronized boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
    if (null == group || null == producer) {
        return false;
    }
    MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
    if (prev != null) {
        log.warn("the producer group[{}] exist already.", group);
        return false;
    }
    return true;
}

更改serviceState状态为运行中

this.serviceState = ServiceState.RUNNING;

开启定时线程池扫描过期请求

入口：org.apache.rocketmq.client.producer.RequestFutureHolder#startScheduledTask

public synchronized void startScheduledTask(DefaultMQProducerImpl producer) {
    this.producerSet.add(producer);
    if (null == scheduledExecutorService) {
        this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("RequestHouseKeepingService"));
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    RequestFutureHolder.getInstance().scanExpiredRequest();
                } catch (Throwable e) {
                    log.error("scan RequestFutureTable exception", e);
                }
            }
        }, 1000 * 3, 1000, TimeUnit.MILLISECONDS);
    }
}