如何自己做网站优化,网站建设及维护服务,广州市城市建设,松江区做网站的公司文章目录 初始化DefaultMQProducer实例启动流程DefaultMQProducer#startDefaultMQProducerImpl#startMQClientInstance#start启动流程总结 实例内容 初始化DefaultMQProducer实例
初始化一个 DefaultMQProducer 对象的代码如下
// 返回一个producer对象
DefaultMQProducer pr… 文章目录 初始化DefaultMQProducer实例启动流程DefaultMQProducer#startDefaultMQProducerImpl#startMQClientInstance#start启动流程总结 实例内容 初始化DefaultMQProducer实例
初始化一个 DefaultMQProducer 对象的代码如下
// 返回一个producer对象
DefaultMQProducer producer new DefaultMQProducer();
// 设置组名
producer.setProducerGroup(ProducerGroupName);
// 设置NameServer地址
producer.setNamesrvAddr(127.0.0.1:9876);在初始化 DefaultMQProducer 时会初始化一个 DefaultMQProducerImpl 实例并赋值给 producer 的成员变量
public DefaultMQProducer() {this(null, MixAll.DEFAULT_PRODUCER_GROUP, null);
}public DefaultMQProducer(final String namespace, final String producerGroup, RPCHook rpcHook) {this.namespace namespace;this.producerGroup producerGroup;// 将defaultMQProducerImpl对象保存在成员变量中defaultMQProducerImpl new DefaultMQProducerImpl(this, rpcHook);
}同时在初始化 DefaultMQProducerImpl 实例时也会将 producer 对象作为成员变量保存在 DefaultMQProducerImpl 实例中
构造 defaultMQProducerImpl 的代码如下
public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) {// 将defaultMQProducer对象保存在成员变量中this.defaultMQProducer defaultMQProducer;this.rpcHook rpcHook;this.asyncSenderThreadPoolQueue new LinkedBlockingQueue(50000);this.defaultAsyncSenderExecutor new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),Runtime.getRuntime().availableProcessors(),1000 * 60,TimeUnit.MILLISECONDS,this.asyncSenderThreadPoolQueue,new ThreadFactory() {private AtomicInteger threadIndex new AtomicInteger(0);Overridepublic Thread newThread(Runnable r) {return new Thread(r, AsyncSenderExecutor_ this.threadIndex.incrementAndGet());}});if (defaultMQProducer.getBackPressureForAsyncSendNum() 10) {semaphoreAsyncSendNum new Semaphore(Math.max(defaultMQProducer.getBackPressureForAsyncSendNum(),10), true);} else {semaphoreAsyncSendNum new Semaphore(10, true);log.info(semaphoreAsyncSendNum can not be smaller than 10.);}if (defaultMQProducer.getBackPressureForAsyncSendNum() 1024 * 1024) {semaphoreAsyncSendSize new Semaphore(Math.max(defaultMQProducer.getBackPressureForAsyncSendNum(),1024 * 1024), true);} else {semaphoreAsyncSendSize new Semaphore(1024 * 1024, true);log.info(semaphoreAsyncSendSize can not be smaller than 1M.);}
}因此 DefaultMQProducerImpl 对象能够通过其保存的 producer 对象实例获取到 producer 的所有参数
经过上面简单的设置后此时 producer 被赋值的成员变量有
producerGroup ProducerGroupName
namesrvAddr 127.0.0.1:9876
namespaceInitialized false
defaultMQProducerImpl new DefaultMQProducerImpl(this, rpcHook) // 这里rpcHook为null启动流程
DefaultMQProducer#start
只需要执行下面代码即可启动上方初始化的 produer 对象
producer.start();producer 的 start() 方法具体的内容如下
public void start() throws MQClientException {// 由于DefaultMQProducer继承了ClientConfig所以可以直接使用ClientConfig的withNamespace方法this.setProducerGroup(withNamespace(this.producerGroup));// 调用DefaultMQProducerImpl的start方法this.defaultMQProducerImpl.start();// 如果traceDispatcher不为空则调用traceDispatcher的start方法if (null ! traceDispatcher) {try {traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());} catch (MQClientException e) {logger.warn(trace dispatcher start failed , e);}}
}可以看到 DefaultMQProducer 只是一个门面类具体的实现都是由DefaultMQProducerImpl 去做的 由于 Namespace 的存在因此在启动 producer 时首先会重新设置 producerGroup我们需要重点关注经过 withNamespace() 方法处理后返回的生产者组名
public String withNamespace(String resource) {// this.getNamespace()不设置的话返回的是nullreturn NamespaceUtil.wrapNamespace(this.getNamespace(), resource);
}可见 withNamespace() 方法仅仅是调用了 wrapNamespace() 方法并将 Namespace 和 producerGroup 作为参数一并传入
public static String wrapNamespace(String namespace, String resourceWithOutNamespace) {// 如果namespace为空或者resourceWithOutNamespace为空则直接返回resourceWithOutNamespaceif (StringUtils.isEmpty(namespace) || StringUtils.isEmpty(resourceWithOutNamespace)) {return resourceWithOutNamespace;}// 如果resourceWithOutNamespace是SystemResource// 或者resourceWithOutNamespace已经组合了Namespace则直接返回resourceWithOutNamespaceif (isSystemResource(resourceWithOutNamespace) || isAlreadyWithNamespace(resourceWithOutNamespace, namespace)) {return resourceWithOutNamespace;}String resourceWithoutRetryAndDLQ withOutRetryAndDLQ(resourceWithOutNamespace);StringBuilder stringBuilder new StringBuilder();if (isRetryTopic(resourceWithOutNamespace)) {stringBuilder.append(MixAll.RETRY_GROUP_TOPIC_PREFIX);}if (isDLQTopic(resourceWithOutNamespace)) {stringBuilder.append(MixAll.DLQ_GROUP_TOPIC_PREFIX);}// 返回 [RETRY_PREFIX] [DLQ_PREFIX] namespace % resourceWithoutRetryAndDLQreturn stringBuilder.append(namespace).append(NAMESPACE_SEPARATOR).append(resourceWithoutRetryAndDLQ).toString();}由于我们并没有设置 producer 的 Namespace因此会直接返回 producerGroup。造成的效果就是在这个生产者启动过程中第一行代码没有任何效果 traceDispatcher 又是个什么东西traceDispatcher 的作用是追踪消息的发送和消费的轨迹它是一个 AsyncTraceDispatcher 对象它实现了 TraceDispatcher 接口用于异步地发送追踪消息到 Broker。它可以帮助用户查看每条消息的完整链路数据包括发送时间、消费时间、存储时间等。我们可以通过使用下面的构造函数构造出一个含有 traceDispatcher 的 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 trace featureif (enableMsgTrace) {try {AsyncTraceDispatcher dispatcher new AsyncTraceDispatcher(producerGroup, TraceDispatcher.Type.PRODUCE, customizedTraceTopic, rpcHook);dispatcher.setHostProducer(this.defaultMQProducerImpl);traceDispatcher dispatcher;this.defaultMQProducerImpl.registerSendMessageHook(new SendMessageTraceHookImpl(traceDispatcher));this.defaultMQProducerImpl.registerEndTransactionHook(new EndTransactionTraceHookImpl(traceDispatcher));} catch (Throwable e) {logger.error(system mqtrace hook init failed ,maybe cant send msg trace data);}}
}由于我们在初始化 DefaultMQProducer 实例时没有生成 traceDispatcher 实例因此 null ! traceDispatcher 返回 FALSE不调用 traceDispatcher 的 start 方法
DefaultMQProducerImpl#start
接下来调用 defaultMQProducerImpl 的 start() 方法方法内容如下
private ServiceState serviceState ServiceState.CREATE_JUST;public void start() throws MQClientException {this.start(true);
}public void start(final boolean startFactory) throws MQClientException {// serviceState默认为CREATE_JUSTswitch (this.serviceState) {case CREATE_JUST:// 1. 设置serviceState为START_FAILEDthis.serviceState ServiceState.START_FAILED;// 2. 检查生产者组名是否合法this.checkConfig();// 3. 如果实例名为默认值则将生产者的instanceName设置为 UtilAll.getPid() # System.nanoTime()if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {this.defaultMQProducer.changeInstanceNameToPID();}// 4. 使用MQClientManager.getInstance()返回一个单例的MQClientManager对象// defaultMQProducer继承了ClientConfig因此getOrCreateMQClientInstance方法的参数可以是defaultMQProducer// mQClientFactory是MQClientInstance的一个实例MQClientInstance是MQClientManager的内部类this.mQClientFactory MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);// 5. 注册producer实例将生产者组名作为keydefaultMQProducerImpl对象作为value保存到MQClientInstance的producerTable中boolean registerOK mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);if (!registerOK) {// 如果注册失败则将serviceState重设为CREATE_JUSTthis.serviceState ServiceState.CREATE_JUST;throw new MQClientException(The producer group[ this.defaultMQProducer.getProducerGroup() ] has been created before, specify another name please. FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),null);}// 6. 将defaultMQProducer的createTopicKey作为keyTopicPublishInfo作为value放入到defaultMQProducerImpl的topicPublishInfoTable中// createTopicKey的默认值为TBW102// topicPublishInfoTable的作用是存储topic的路由信息包括topic的queue数目、brokerName、brokerId等this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());// 7. 启动MQClientInstance实例if (startFactory) {// MQClientInstance的start方法会启动MQClientInstance的定时任务// 包括定时向所有broker发送心跳、定时清理过期的topic、定时清理过期的consumer、定时清理过期的producermQClientFactory.start();}log.info(the producer [{}] start OK. sendMessageWithVIPChannel{}, this.defaultMQProducer.getProducerGroup(),this.defaultMQProducer.isSendMessageWithVIPChannel());// 8. 如果启动成功则将serviceState设置为RUNNINGthis.serviceState ServiceState.RUNNING;break;case RUNNING:case START_FAILED:case SHUTDOWN_ALREADY:throw new MQClientException(The producer service state not OK, maybe started once, this.serviceState FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),null);default:break;}// 立即发送心跳到所有brokerthis.mQClientFactory.sendHeartbeatToAllBrokerWithLock();// 开启一个定时任务来处理所有的 Request 状态对异步的请求根据状态处理回调函数// 这个异步请求指的并不是在 send 中的异步回调机制而是 Request-Reply 特性用来模拟 RPC 调用RequestFutureHolder.getInstance().startScheduledTask(this);}注意上述代码中有一段注释为createTopicKey 的默认值为 TBW102这个 Topic 在自动创建 topic 时有关键作用
最后的 RequestFutureHolder.getInstance().startScheduledTask(this) 用来扫描和处理过期的异步请求但是需要注意的是这个异步请求指的并不是在 send 中的异步回调机制而是 Request-Reply 特性用来模拟 RPC 调用 RocketMQ 有两种异步请求的方式一种是在 send 方法中传入一个回调函数当消息发送成功或失败时会调用这个回调函数。这种方式不需要等待服务器的响应只需要等待服务器的确认。 另一种是在 RocketMQ 4.7.0 后加入的 Request-Reply 特性这种方式是模拟 RPC 调用需要等待服务器的响应并返回一个结果。这种方式需要使用 RequestResponseFuture 对象来封装请求和响应的信息。 在使用 getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook) 返回一个 MQClientInstance 对象时如果 factoryTable 中没有实例的话则会初始化一个新的实例代码如下
public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {// 最多情况下 clientId ipinstanceNameunitNameRequestType.STREAM// 默认情况下 clientId ipinstanceNameString 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);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 实例的过程中会初始化一个新的 DefaultMQProducer 实例与我们一开始就有的 producer 实例不是一个对象
public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {// ...// 此处实例化内部的producer// 用于消费失败或超时的消息sendMessageBack回发给broker放到retry topic中重试消费this.defaultMQProducer new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);this.defaultMQProducer.resetClientConfig(clientConfig);// ...
}此时新初始化的 DefaultMQProducer 实例的 producerGroup CLIENT_INNER_PRODUCERinstanceName DEFAULT
接着在初始化实例后又执行了 this.defaultMQProducer.resetClientConfig(clientConfig) 这行代码
public void resetClientConfig(final ClientConfig cc) {this.namesrvAddr cc.namesrvAddr;this.clientIP cc.clientIP;this.instanceName cc.instanceName;this.clientCallbackExecutorThreads cc.clientCallbackExecutorThreads;this.pollNameServerInterval cc.pollNameServerInterval;this.heartbeatBrokerInterval cc.heartbeatBrokerInterval;this.persistConsumerOffsetInterval cc.persistConsumerOffsetInterval;this.pullTimeDelayMillsWhenException cc.pullTimeDelayMillsWhenException;this.unitMode cc.unitMode;this.unitName cc.unitName;this.vipChannelEnabled cc.vipChannelEnabled;this.useTLS cc.useTLS;this.socksProxyConfig cc.socksProxyConfig;this.namespace cc.namespace;this.language cc.language;this.mqClientApiTimeout cc.mqClientApiTimeout;this.decodeReadBody cc.decodeReadBody;this.decodeDecompressBody cc.decodeDecompressBody;this.enableStreamRequestType cc.enableStreamRequestType;
}可以看到这段代码将 producer 实例的 ClinetConfig 属性完全拷贝了一份给新创建的 DefaultMQProducer 实例的 ClinetConfig 属性。因此这段代码后新初始化的 DefaultMQProducer 实例的 instanceName 不再是默认值而是和 producer 的一致
MQClientInstance#start
接下来我们来看 mQClientFactory.start() 这部分的代码
public void start() throws MQClientException {// 用synchronized修饰保证线程安全性与内存可见性synchronized (this) {switch (this.serviceState) {case CREATE_JUST:this.serviceState ServiceState.START_FAILED;// If not specified,looking address from name server// 由于传入了NameServer的地址因此不进入分支if (null this.clientConfig.getNamesrvAddr()) {this.mQClientAPIImpl.fetchNameServerAddr();}// Start request-response channel// 1. 启动用于和broker通信的netty客户端this.mQClientAPIImpl.start();// Start various schedule tasks// 2. 启动定时任务包括心跳拉取topic路由信息更新broker信息清理过期消息等this.startScheduledTask();// 3. Start pull servicethis.pullMessageService.start();// Start rebalance service// 4. 启动负载均衡服务对MQConsumer有效this.rebalanceService.start();// Start push service// 5. 启动它内部的producer实例// this.defaultMQProducer是在DefaultMQProducerImpl中// 使用MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook)被初始化的this.defaultMQProducer.getDefaultMQProducerImpl().start(false);log.info(the client factory [{}] start OK, this.clientId);this.serviceState ServiceState.RUNNING;break;case START_FAILED:throw new MQClientException(The Factory object[ this.getClientId() ] has been created before, and failed., null);default:break;}}
}这段代码的主要任务是启动和broker通信的 Netty 客户端、启动定时任务、启动拉取消息服务、启动负载均衡服务、启动内部生产者实例
其中启动负载均衡服务只是针对消费者而言的在生产者启动过程中并无作用
mQClientAPIImpl 对象是 RocketMQ 客户端与 Broker 之间通信的实现类
我们先看启动定时任务这个方法做了什么
private void startScheduledTask() {if (null this.clientConfig.getNamesrvAddr()) {// 如果没有指定namesrv地址则定时获取namesrv地址this.scheduledExecutorService.scheduleAtFixedRate(() - {try {MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();} catch (Exception e) {log.error(ScheduledTask fetchNameServerAddr exception, e);}}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);}// 定期从NameServer更新topic路由信息this.scheduledExecutorService.scheduleAtFixedRate(() - {try {MQClientInstance.this.updateTopicRouteInfoFromNameServer();} catch (Exception e) {log.error(ScheduledTask updateTopicRouteInfoFromNameServer exception, e);}}, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);// 用于定期清除离线Broker并向所有Broker发送心跳包this.scheduledExecutorService.scheduleAtFixedRate(() - {try {MQClientInstance.this.cleanOfflineBroker();MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();} catch (Exception e) {log.error(ScheduledTask sendHeartbeatToAllBroker exception, e);}}, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);// 定时持久化消费者当前消费进度对MQConsumer有效this.scheduledExecutorService.scheduleAtFixedRate(() - {try {MQClientInstance.this.persistAllConsumerOffset();} catch (Exception e) {log.error(ScheduledTask persistAllConsumerOffset exception, e);}}, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);// 根据当前的积压调优线程池的核心线程数this.scheduledExecutorService.scheduleAtFixedRate(() - {try {MQClientInstance.this.adjustThreadPool();} catch (Exception e) {log.error(ScheduledTask adjustThreadPool exception, e);}}, 1, 1, TimeUnit.MINUTES);
}其中的定时持久化消费者当前消费进度任务同样只是针对消费者而言的在生产者启动过程中并无作用
至此producer 的启动流程结束
启动流程总结
执行 producer.start()启动一个 producer
重新设置生产者组名调用 defaultMQProducerImpl 的 start() 方法是进行启动实现的入口 检查当前状态如果是 CREATE_JUST 则进入启动流程检查生产者组名称是否合法更改 producer 的 instanceName创建一个 MQClientInstance 类型的 mQClinetFactory 实例 创建一个新的 defaultMQProducer 实例作为 mQClinetFactory 实例的成员变量新的 defaultMQProducer 实例内部又会创建一个新的 defaultMQProducerImpl 实例将新的 defaultMQProducer 实例的 ClinetConfig 属性复制粘贴为 producer 的 ClinetConfig 属性 将 producer 实例放入 mQClinetFactory 的 producerTable 中key为 producer 的生产者组名将 defaultMQProducer 的 createTopicKey 作为keyTopicPublishInfo 作为value放入到 defaultMQProducerImpl 的 topicPublishInfoTable 中启动 mQClinetFactory 如果没有 NameServer 地址则尝试获取启动用于和 broker 通信的 netty 客户端启动定时任务 如果没有指定 NameServer 地址则定时获取 NameServer 地址定期从NameServer更新topic路由信息定期清除离线Broker并向所有Broker发送心跳包定时持久化消费者当前消费进度对MQConsumer有效定时根据当前的积压调优线程池的核心线程数但是实现是空的 启动 pullMessageService 从 broker 拉取消息启动消费者客户端的负载均衡服务启动 mQClinetFactory 内部的 defaultMQProducerImpl 实例 检查当前状态如果是 CREATE_JUST 则进入启动流程检查生产者组名称是否合法由于其 instanceName 等于 MixAll.CLIENT_INNER_PRODUCER_GROUP因此不更改创建一个 MQClientInstance 类型的 mQClinetFactory 实例将 producer 实例放入 mQClinetFactory 的 producerTable 中key为 producer 的生产者组名将 defaultMQProducer 的 createTopicKey 作为keyTopicPublishInfo 作为value放入到 defaultMQProducerImpl 的 topicPublishInfoTable 中将当前状态设置为 RUNNING如果上述都成功则立即发送心跳到所有的 broker启动定时任务扫描和处理过期的异步请求 将当前状态设置为 RUNNING如果上述都成功则立即发送心跳到所有的 broker启动定时任务扫描和处理过期的异步请求 如果 traceDispatch 不为空则启动 traceDispatcher
实例内容
以生产者组名称为 ProducerGroupNametopic名称为 TopicTest 为例生产者启动成功后的实例内容如下
DefaultMQProducer producer:1 # 启动入口String namesrvAddr 127.0.0.1:9876 # NameServer地址String clientIp 192.168.142.1 # producer的ip地址String producerGroup ProducerGroupName # 生产者组名称String createTopicKey TBW102 # 用来自动创建topic的topic名称String instanceName 14896#9822706678400 # 生产者实例的名称DefaultProducerImpl defaultProducerImpl:1 # 实际启动类DefaultMQProducer defaultMQProducer:1 # 就是最开头的 producerConcurrentMapString/* topic */, TopicPublishInfo topicPublishInfoTable:1 {TBW102 : TopicPublishInfo} # 存储topic的路由信息MQClientInstance mQClientFactory # 负责管理与 NameServer 和 Broker 的网络连接String clientId 192.168.142.114896#9822706678400 # 为 ip instanceNameMQClientAPIImpl mQClientAPIImpl # 负责实现与 Broker 之间的通信ConcurrentMapString, MQProducerInner producerTable {ProducerGroupName : defaultProducerImpl:1} # MQProducerInner是DefaultProducerImpl的接口ConcurrentMapString, MQConsumerInner consumerTable # MQConsumerInner是DefaultConsumerImpl的接口ConcurrentMapString, HashMapLong, String brokerAddrTable # broker的地址DefaultMQProducer defaultMQProducer:2 # mQClientFactory内部的producer实例String producerGroup CLIENT_INNER_PRODUCERStirng instanceName 14896#9822706678400 # 和 producer 的 clientConfig 属性完全一致DefaultProducerImpl defaultProducerImpl:2DefaultMQProducer defaultMQProducer:2ConcurrentMapString/* topic */, TopicPublishInfo topicPublishInfoTable:2 {TBW102 : TopicPublishInfo} # 存储topic的路由信息
