RocketMQ中Producer的启动源码分析
RocketMQ中通过DefaultMQProducer创建Producer
DefaultMQProducer定义如下:
public class DefaultMQProducer extends ClientConfig implements MQProducer { protected final transient DefaultMQProducerImpl defaultMQProducerImpl; private String createTopicKey = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC = "TBW102"; private volatile int defaultTopicQueueNums = 4; private int sendMsgTimeout = 3000; private int compressMsgBodyOverHowmuch = 1024 * 4; private int retryTimesWhenSendFailed = 2; private int retryTimesWhenSendAsyncFailed = 2; private boolean retryAnotherBrokerWhenNotStoreOK = false; private int maxMessageSize = 1024 * 1024 * 4; // 4M }
其中defaultMQProducerImpl成员是Producer的具体实现,其余的一些成员是对一些参数的设置:
createTopicKey:是一个Topic值,在创建时使用,后面会说明
defaultTopicQueueNums :默认的Topic队列个数
sendMsgTimeout:发送消息超时时间
compressMsgBodyOverHowmuch:消息容量限制,超过需要进行压缩
retryTimesWhenSendFailed:同步消息发送失败的允许重发次数
retryTimesWhenSendAsyncFailed:异步消息发送失败的允许重发次数
retryAnotherBrokerWhenNotStoreOK:是否允许发送给Broker失败后,重新选择Broker发送
maxMessageSize:消息最大大小
这些属性可以通过DefaultMQProducer提供的get、set方法进行相应操作
常用的构造方法如下:
public DefaultMQProducer() { this(MixAll.DEFAULT_PRODUCER_GROUP, null); } public DefaultMQProducer(final String producerGroup) { this(producerGroup, null); } public DefaultMQProducer(final String producerGroup, RPCHook rpcHook) { this.producerGroup = producerGroup; defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook); }
DefaultMQProducer继承自ClientConfig,首先会设置ClientConfig提供的更底层的参数配置:
public class ClientConfig { public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel"; private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV)); private String clientIP = RemotingUtil.getLocalAddress(); private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT"); private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors(); private int pollNameServerInterval = 1000 * 30; private int heartbeatBrokerInterval = 1000 * 30; private boolean vipChannelEnabled = Boolean.parseBoolean(System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "true")); }
其中namesrvAddr是非常重要的成员,其保存着名称服务器(Name Server)的地址,在一开始构造时会根据系统属性进行设置,若是没有设置系统属性就是null,则需要在后面通过set方法进行设置
clientIP:Producer端的本地IP
instanceName:Producer的实例名称
pollNameServerInterval :轮询NameServer的时间间隔
heartbeatBrokerInterval :向Broker发送心跳包的时间间隔
SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY 和vipChannelEnabled:决定是否使用VIP通道,即高优先级
回到DefaultMQProducer的构造方法,其会创建DefaultMQProducerImpl实例
private final Random random = new Random(); private final ConcurrentMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable = new ConcurrentHashMap<String, TopicPublishInfo>(); private final ArrayList<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>(); private final RPCHook rpcHook; protected BlockingQueue<Runnable> checkRequestQueue; protected ExecutorService checkExecutor; private ServiceState serviceState = ServiceState.CREATE_JUST; private MQClientInstance mQClientFactory; private ArrayList<CheckForbiddenHook> checkForbiddenHookList = new ArrayList<CheckForbiddenHook>(); private int zipCompressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5")); private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy(); private final BlockingQueue<Runnable> asyncSenderThreadPoolQueue; private final ExecutorService defaultAsyncSenderExecutor; private ExecutorService asyncSenderExecutor; public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) { this.defaultMQProducer = defaultMQProducer; this.rpcHook = rpcHook; this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(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); @Override public Thread newThread(Runnable r) { return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet()); } }); }
在构造方法中会创建一个线程池,用来处理异步消息的发送
其中有一个topicPublishInfoTable成员很重要,是一个map,保存了不同top和消息队列之间的映射,在后面详细介绍
DefaultMQProducer创建完成后,接着来看DefaultMQProducer的start方法:
public void start() throws MQClientException { this.defaultMQProducerImpl.start(); if (null != traceDispatcher) { try { traceDispatcher.start(this.getNamesrvAddr()); } catch (MQClientException e) { log.warn("trace dispatcher start failed ", e); } } }
首先交给了defaultMQProducerImpl的start方法去处理
defaultMQProducerImpl的start方法:
public void start() throws MQClientException { this.start(true); } public void start(final boolean startFactory) throws MQClientException { switch (this.serviceState) { case CREATE_JUST: this.serviceState = ServiceState.START_FAILED; this.checkConfig(); if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) { this.defaultMQProducer.changeInstanceNameToPID(); } this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook); boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this); if (!registerOK) { this.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); } this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo()); if (startFactory) { mQClientFactory.start(); } log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(), this.defaultMQProducer.isSendMessageWithVIPChannel()); this.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; } this.mQClientFactory.sendHeartbeatToAllBrokerWithLock(); }
在一开始DefaultMQProducerImpl实例化的时候,serviceState初始化为CREATE_JUST状态,这是一个枚举值,一共有如下几种状态:
public enum ServiceState { CREATE_JUST, RUNNING, SHUTDOWN_ALREADY, START_FAILED; private ServiceState() { } }
这几个状态值很容易理解,在后面MQClientInstance中还会使用到
回到start方法,根据serviceState进行判断,只有当是CREATE_JUST状态时正常执行,防止在其他状态下错误调用start
直接看到CREATE_JUST的case部分:
this.serviceState = ServiceState.START_FAILED; this.checkConfig(); if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) { this.defaultMQProducer.changeInstanceNameToPID(); } this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook); boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this); if (!registerOK) { this.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); } this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo()); if (startFactory) { mQClientFactory.start(); } log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(), this.defaultMQProducer.isSendMessageWithVIPChannel()); this.serviceState = ServiceState.RUNNING; break;
首先更改serviceState状态为START_FAILED,防止中途的失败
checkConfig方法是用来进行ProducerGroup命名检查:
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); } }
主要是检查命名的合法性,以及防止和默认的producerGroup生产者组名DEFAULT_PRODUCER_GROUP产生冲突
public static final String DEFAULT_PRODUCER_GROUP = "DEFAULT_PRODUCER";
接下来实例化mQClientFactory,这其实是生产者客户端的实例,其中MQClientManager采用单例模式,getInstance是获取MQClientManager的单例,根据ClientConfig的类型,通过getAndCreateMQClientInstance方法实例化不同属性的生产者客户端
MQClientManager:
public class MQClientManager { private final static InternalLogger log = ClientLogger.getLog(); private static MQClientManager instance = new MQClientManager(); private AtomicInteger factoryIndexGenerator = new AtomicInteger(); private ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable = new ConcurrentHashMap<String, MQClientInstance>(); private MQClientManager() { } public static MQClientManager getInstance() { return instance; } }
其中factoryTable是所有生产者客户端实例的map缓存,factoryIndexGenerator 是创建的每个客户端实例的流水号
getAndCreateMQClientInstance方法:
public MQClientInstance getAndCreateMQClientInstance(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); if (prev != null) { instance = prev; log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId); } else { log.info("Created new MQClientInstance for clientId:[{}]", clientId); } } return instance; }
首先通过buildMQClientId方法创建clientId:
public String buildMQClientId() { StringBuilder sb = new StringBuilder(); sb.append(this.getClientIP()); sb.append("@"); sb.append(this.getInstanceName()); if (!UtilAll.isBlank(this.unitName)) { sb.append("@"); sb.append(this.unitName); } return sb.toString(); }
clientId主要由生产者客户端的ip地址以及实例名称,根据unitName的有无,附加unitName
通过生成的clientId,在factoryTable缓存中先去获取是否创建过客户端实例
若是没有获取到,就需要实例化一个MQClientInstance
这里在实例化MQClientInstance时,并没有直接传入clientConfig,而是通过cloneClientConfig方法复制了一份,来保证安全性:
public ClientConfig cloneClientConfig() { ClientConfig cc = new ClientConfig(); cc.namesrvAddr = namesrvAddr; cc.clientIP = clientIP; cc.instanceName = instanceName; cc.clientCallbackExecutorThreads = clientCallbackExecutorThreads; cc.pollNameServerInterval = pollNameServerInterval; cc.heartbeatBrokerInterval = heartbeatBrokerInterval; cc.persistConsumerOffsetInterval = persistConsumerOffsetInterval; cc.unitMode = unitMode; cc.unitName = unitName; cc.vipChannelEnabled = vipChannelEnabled; cc.useTLS = useTLS; cc.language = language; return cc; }
创建MQClientInstance实例:
public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) { this.clientConfig = clientConfig; this.instanceIndex = instanceIndex; this.nettyClientConfig = new NettyClientConfig(); this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads()); this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS()); this.clientRemotingProcessor = new ClientRemotingProcessor(this); this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig); if (this.clientConfig.getNamesrvAddr() != null) { this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr()); log.info("user specified name server address: {}", this.clientConfig.getNamesrvAddr()); } this.clientId = clientId; this.mQAdminImpl = new MQAdminImpl(this); this.pullMessageService = new PullMessageService(this); this.rebalanceService = new RebalanceService(this); this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP); this.defaultMQProducer.resetClientConfig(clientConfig); this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService); log.info("Created a new client Instance, InstanceIndex:{}, ClientID:{}, ClientConfig:{}, ClientVersion:{}, SerializerType:{}", this.instanceIndex, this.clientId, this.clientConfig, MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION), RemotingCommand.getSerializeTypeConfigInThisServer()); }
可以看到MQClientInstance的构造方法创建了很多东西,就不一一说明,主要说几个重要的
其中nettyClientConfig,就很清楚的说明了RocketMQ通过Netty来进行网络之间的I/O,其保存了对Netty的一些配置
clientRemotingProcessor,用来进行消息的处理
mQClientAPIImpl则是一个非常重要的部分,直接实例化了一个MQClientAPIImpl对象:
public MQClientAPIImpl(final NettyClientConfig nettyClientConfig, final ClientRemotingProcessor clientRemotingProcessor, RPCHook rpcHook, final ClientConfig clientConfig) { this.clientConfig = clientConfig; topAddressing = new TopAddressing(MixAll.getWSAddr(), clientConfig.getUnitName()); this.remotingClient = new NettyRemotingClient(nettyClientConfig, null); this.clientRemotingProcessor = clientRemotingProcessor; this.remotingClient.registerRPCHook(rpcHook); this.remotingClient.registerProcessor(RequestCode.CHECK_TRANSACTION_STATE, this.clientRemotingProcessor, null); this.remotingClient.registerProcessor(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, this.clientRemotingProcessor, null); this.remotingClient.registerProcessor(RequestCode.RESET_CONSUMER_CLIENT_OFFSET, this.clientRemotingProcessor, null); this.remotingClient.registerProcessor(RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT, this.clientRemotingProcessor, null); this.remotingClient.registerProcessor(RequestCode.GET_CONSUMER_RUNNING_INFO, this.clientRemotingProcessor, null); this.remotingClient.registerProcessor(RequestCode.CONSUME_MESSAGE_DIRECTLY, this.clientRemotingProcessor, null); }
可以看到在这个构造方法里,首先创建了一个TopAddressing,用于以后的名称服务的寻址,其默认地址是:
http://jmenv.tbsite.net:8080/rocketmq/nsaddr
需要通过系统属性来完成更改
接着创建了一个NettyRemotingClient,这个就是实实在在的Netty客户端
private final Bootstrap bootstrap = new Bootstrap(); // 名称服务列表 private final AtomicReference<List<String>> namesrvAddrList = new AtomicReference<List<String>>(); public NettyRemotingClient(final NettyClientConfig nettyClientConfig, final ChannelEventListener channelEventListener) { super(nettyClientConfig.getClientOnewaySemaphoreValue(), nettyClientConfig.getClientAsyncSemaphoreValue()); this.nettyClientConfig = nettyClientConfig; this.channelEventListener = channelEventListener; int publicThreadNums = nettyClientConfig.getClientCallbackExecutorThreads(); if (publicThreadNums <= 0) { publicThreadNums = 4; } this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() { private AtomicInteger threadIndex = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { return new Thread(r, "NettyClientPublicExecutor_" + this.threadIndex.incrementAndGet()); } }); this.eventLoopGroupWorker = new NioEventLoopGroup(1, new ThreadFactory() { private AtomicInteger threadIndex = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { return new Thread(r, String.format("NettyClientSelector_%d", this.threadIndex.incrementAndGet())); } }); if (nettyClientConfig.isUseTLS()) { try { sslContext = TlsHelper.buildSslContext(true); log.info("SSL enabled for client"); } catch (IOException e) { log.error("Failed to create SSLContext", e); } catch (CertificateException e) { log.error("Failed to create SSLContext", e); throw new RuntimeException("Failed to create SSLContext", e); } } }
此时Netty的客户端仅仅完成了对Bootstrap的初始化,以及对NioEventLoopGroup的设置和初始化
回到MQClientInstance的构造方法,在完成MQClientAPIImpl的创建后,会根据clientConfig的getNamesrvAddr判断是否设置了namesrvAddr名称服务地址,若是设置了,需要通过mQClientAPIImpl的updateNameServerAddressList方法,完成对名称服务地址的更新:
MQClientAPIImpl的updateNameServerAddressList方法:
public void updateNameServerAddressList(final String addrs) { String[] addrArray = addrs.split(";"); List<String> list = Arrays.asList(addrArray); this.remotingClient.updateNameServerAddressList(list); }
由于名称服务可以是集群的方式,所以在这里用“;”进行分割,得到所有的名称服务地址,再由remotingClient进行更新,而此时的remotingClient也就是刚才创建的NettyRemotingClient
NettyRemotingClient的updateNameServerAddressList方法:
public void updateNameServerAddressList(List<String> addrs) { List<String> old = this.namesrvAddrList.get(); boolean update = false; if (!addrs.isEmpty()) { if (null == old) { update = true; } else if (addrs.size() != old.size()) { update = true; } else { for (int i = 0; i < addrs.size() && !update; i++) { if (!old.contains(addrs.get(i))) { update = true; } } } if (update) { Collections.shuffle(addrs); log.info("name server address updated. NEW : {} , OLD: {}", addrs, old); this.namesrvAddrList.set(addrs); } } }
这里逻辑比较简单,完成了名称服务列表的更新
回到MQClientInstance的构造方法,做完以上操作后,又在后面创建了MQAdminImpl、PullMessageService、RebalanceService、ConsumerStatsManager以及一个新的DefaultMQProducer,关于这几个在后面出现时再介绍
回到MQClientManager的getAndCreateMQClientInstance方法,在完成MQClientInstance的创建后,将其放入缓存中
再回到DefaultMQProducerImpl的start方法,在创建完MQClientInstance后,调用registerProducer方法
MQClientInstance的registerProducer方法:
public 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; }
在MQClientInstance初始化时,会创建producerTable 、consumerTable 、topicRouteTable 、brokerAddrTable 这几个比较重要的map
private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>(); private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>(); private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>(); private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable = new ConcurrentHashMap<String, HashMap<Long, String>>();
其中MQProducerInner是接口,DefaultMQProducerImpl是其实现类,完成了以group组名称为键值的DefaultMQProducerImpl的关联
在这里就是根据group,进行DefaultMQProducerImpl的缓存,MQConsumerInner同理
topicRouteTable 则记录与Topic对应的Broker以及消息队列信息
brokerAddrTable则记录与Broker Name对应的Broker的地址列表
还是回到start方法,在完成registerProducer方法后,根据返回值registerOK,判断接下来的操作
若是失败,将serviceState置为CREATE_JUST,并报出异常,方便下一次的正常start
若是成功,则先需要向topicPublishInfoTable中添加一条键值为createTopicKey("TBW102")的TopicPublishInfo记录
TopicPublishInfo:
public class TopicPublishInfo { private boolean orderTopic = false; private boolean haveTopicRouterInfo = false; private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>(); private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex(); }
其中messageQueueList存放消息队列MessageQueue,sendWhichQueue 是用来获取sendWhichQueue中的下标,也就是当前所要发送的具体的消息队列
MessageQueue:
public class MessageQueue implements Comparable<MessageQueue>, Serializable { private static final long serialVersionUID = 6191200464116433425L; private String topic; private String brokerName; private int queueId; public MessageQueue() { } public MessageQueue(String topic, String brokerName, int queueId) { this.topic = topic; this.brokerName = brokerName; this.queueId = queueId; } }
可以看到这是一个简单的pojo,其封装了topic,brokerName以及queueId
ThreadLocalIndex :
public class ThreadLocalIndex { private final ThreadLocal<Integer> threadLocalIndex = new ThreadLocal<Integer>(); private final Random random = new Random(); public int getAndIncrement() { Integer index = this.threadLocalIndex.get(); if (null == index) { index = Math.abs(random.nextInt()); if (index < 0) index = 0; this.threadLocalIndex.set(index); } index = Math.abs(index + 1); if (index < 0) index = 0; this.threadLocalIndex.set(index); return index; } @Override public String toString() { return "ThreadLocalIndex{" + "threadLocalIndex=" + threadLocalIndex.get() + '}'; } }
通过ThreadLocal,赋予每个线程一个随机值,后面会根据这个随机值通过和messageQueueList的length取余运算,选取一个MessageQueue ,进而选取一条真正的消息队列进行消息发送
再次回到DefaultMQProducerImpl的start方法,在完成createTopicKey的Topic的记录添加后,根据startFactory判断是否需要调用mQClientFactory的start方法,这里默认startFactory是true,就需要调用mQClientFactory的start方法:
MQClientInstance的start方法:
public void start() throws MQClientException { synchronized (this) { switch (this.serviceState) { case CREATE_JUST: this.serviceState = ServiceState.START_FAILED; // If not specified,looking address from name server if (null == this.clientConfig.getNamesrvAddr()) { this.mQClientAPIImpl.fetchNameServerAddr(); } // Start request-response channel this.mQClientAPIImpl.start(); // Start various schedule tasks this.startScheduledTask(); // Start pull service this.pullMessageService.start(); // Start rebalance service this.rebalanceService.start(); // Start push service this.defaultMQProducer.getDefaultMQProducerImpl().start(false); log.info("the client factory [{}] start OK", this.clientId); this.serviceState = ServiceState.RUNNING; break; case RUNNING: break; case SHUTDOWN_ALREADY: break; case START_FAILED: throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null); default: break; } } }
MQClientInstance在创建时其serviceState状态也是CREATE_JUST
这里首先检查名称服务地址是否设置,若是没有设置,则通过MQClientAPIImpl的fetchNameServerAddr方法,尝试自动获取名称服务
MQClientAPIImpl的fetchNameServerAddr方法:
public String fetchNameServerAddr() { try { String addrs = this.topAddressing.fetchNSAddr(); if (addrs != null) { if (!addrs.equals(this.nameSrvAddr)) { log.info("name server address changed, old=" + this.nameSrvAddr + ", new=" + addrs); this.updateNameServerAddressList(addrs); this.nameSrvAddr = addrs; return nameSrvAddr; } } } catch (Exception e) { log.error("fetchNameServerAddr Exception", e); } return nameSrvAddr; }
这里首先根据topAddressing的fetchNSAddr方法获取名称服务地址,若是获取到了,则判断是否需要更新名称服务列表以及原来的nameSrvAddr
topAddressing在前面说过,MQClientAPIImpl构造方法中,创建TopAddressing实例
TopAddressing的fetchNSAddr方法:
public final String fetchNSAddr() { return fetchNSAddr(true, 3000); } public final String fetchNSAddr(boolean verbose, long timeoutMills) { String url = this.wsAddr; try { if (!UtilAll.isBlank(this.unitName)) { url = url + "-" + this.unitName + "?nofix=1"; } HttpTinyClient.HttpResult result = HttpTinyClient.httpGet(url, null, null, "UTF-8", timeoutMills); if (200 == result.code) { String responseStr = result.content; if (responseStr != null) { return clearNewLine(responseStr); } else { log.error("fetch nameserver address is null"); } } else { log.error("fetch nameserver address failed. statusCode=" + result.code); } } catch (IOException e) { if (verbose) { log.error("fetch name server address exception", e); } } if (verbose) { String errorMsg = "connect to " + url + " failed, maybe the domain name " + MixAll.getWSAddr() + " not bind in /etc/hosts"; errorMsg += FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL); log.warn(errorMsg); } return null; }
首先根据wsAddr和unitName创建url,其中wsAddr在前面说过,默认是http://jmenv.tbsite.net:8080/rocketmq/nsaddr,需要通过系统属性来更改
然后通过HttpTinyClient的httpGet方法建立连接,进行GET请求,获取名称地址
HttpTinyClient的httpGet方法:
static public HttpResult httpGet(String url, List<String> headers, List<String> paramValues, String encoding, long readTimeoutMs) throws IOException { String encodedContent = encodingParams(paramValues, encoding); url += (null == encodedContent) ? "" : ("?" + encodedContent); HttpURLConnection conn = null; try { conn = (HttpURLConnection) new URL(url).openConnection(); conn.setRequestMethod("GET"); conn.setConnectTimeout((int) readTimeoutMs); conn.setReadTimeout((int) readTimeoutMs); setHeaders(conn, headers, encoding); conn.connect(); int respCode = conn.getResponseCode(); String resp = null; if (HttpURLConnection.HTTP_OK == respCode) { resp = IOTinyUtils.toString(conn.getInputStream(), encoding); } else { resp = IOTinyUtils.toString(conn.getErrorStream(), encoding); } return new HttpResult(respCode, resp); } finally { if (conn != null) { conn.disconnect(); } } }
这里就通过了JDK原生的HttpURLConnection ,完成了一次指定url的GET请求,返回请求数据,将请求到的数据以及状态码封装为HttpResult,返回给上一级调用,也就是TopAddressing的fetchNSAddr方法中,再调用clearNewLine方法,将状态码为200的数据处理(清除不必要的空客、换行、回车),得到名称地址,最后回到fetchNameServerAddr方法中,完成名称服务列表的更新,至此自动获取名称服务结束
回到MQClientInstance的start方法中:
在确定有名称服务的情况下,首先调用mQClientAPIImpl的start方法:
MQClientAPIImpl的start方法:
public void start() { this.remotingClient.start(); }
这里实际上调用了前面所创建的Nettt客户端的start方法:
NettyRemotingClient的start方法:
public void start() { this.defaultEventExecutorGroup = new DefaultEventExecutorGroup( nettyClientConfig.getClientWorkerThreads(), new ThreadFactory() { private AtomicInteger threadIndex = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { return new Thread(r, "NettyClientWorkerThread_" + this.threadIndex.incrementAndGet()); } }); Bootstrap handler = this.bootstrap.group(this.eventLoopGroupWorker).channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.SO_KEEPALIVE, false) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis()) .option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize()) .option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize()) .handler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); if (nettyClientConfig.isUseTLS()) { if (null != sslContext) { pipeline.addFirst(defaultEventExecutorGroup, "sslHandler", sslContext.newHandler(ch.alloc())); log.info("Prepend SSL handler"); } else { log.warn("Connections are insecure as SSLContext is null!"); } } pipeline.addLast( defaultEventExecutorGroup, new NettyEncoder(), new NettyDecoder(), new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()), new NettyConnectManageHandler(), new NettyClientHandler()); } }); this.timer.scheduleAtFixedRate(new TimerTask() { @Override public void run() { try { NettyRemotingClient.this.scanResponseTable(); } catch (Throwable e) { log.error("scanResponseTable exception", e); } } }, 1000 * 3, 1000); if (this.channelEventListener != null) { this.nettyEventExecutor.start(); } }
这里完成了Bootstrap对前面创建的EventLoopGroup以及handler的绑定
在完成mQClientAPIImpl的start方法后,调用startScheduledTask方法,启动定时任务
startScheduledTask方法:
private void startScheduledTask() { if (null == this.clientConfig.getNamesrvAddr()) { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr(); } catch (Exception e) { log.error("ScheduledTask fetchNameServerAddr exception", e); } } }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS); } this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { MQClientInstance.this.updateTopicRouteInfoFromNameServer(); } catch (Exception e) { log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e); } } }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS); this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { MQClientInstance.this.cleanOfflineBroker(); MQClientInstance.this.sendHeartbeatToAllBrokerWithLock(); } catch (Exception e) { log.error("ScheduledTask sendHeartbeatToAllBroker exception", e); } } }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS); this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { MQClientInstance.this.persistAllConsumerOffset(); } catch (Exception e) { log.error("ScheduledTask persistAllConsumerOffset exception", e); } } }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS); this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { MQClientInstance.this.adjustThreadPool(); } catch (Exception e) { log.error("ScheduledTask adjustThreadPool exception", e); } } }, 1, 1, TimeUnit.MINUTES); }
可以看到,一共设置了五个定时任务
①若是名称服务地址namesrvAddr不存在,则调用前面的fetchNameServerAddr方法,定时更新名称服务
②通过updateTopicRouteInfoFromNameServer方法定时更新Topic所对应的路由信息:
public void updateTopicRouteInfoFromNameServer() { Set<String> topicList = new HashSet<String>(); // Consumer { Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator(); while (it.hasNext()) { Entry<String, MQConsumerInner> entry = it.next(); MQConsumerInner impl = entry.getValue(); if (impl != null) { Set<SubscriptionData> subList = impl.subscriptions(); if (subList != null) { for (SubscriptionData subData : subList) { topicList.add(subData.getTopic()); } } } } } // Producer { Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator(); while (it.hasNext()) { Entry<String, MQProducerInner> entry = it.next(); MQProducerInner impl = entry.getValue(); if (impl != null) { Set<String> lst = impl.getPublishTopicList(); topicList.addAll(lst); } } } for (String topic : topicList) { this.updateTopicRouteInfoFromNameServer(topic); } }
将所有Consumer和Producer的Topic封装在topicList,交给updateTopicRouteInfoFromNameServer调用
updateTopicRouteInfoFromNameServer方法:
public boolean updateTopicRouteInfoFromNameServer(final String topic) { return updateTopicRouteInfoFromNameServer(topic, false, null); } public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault, DefaultMQProducer defaultMQProducer) { try { if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) { try { TopicRouteData topicRouteData; if (isDefault && defaultMQProducer != null) { topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(), 1000 * 3); if (topicRouteData != null) { for (QueueData data : topicRouteData.getQueueDatas()) { int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums()); data.setReadQueueNums(queueNums); data.setWriteQueueNums(queueNums); } } } else { topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3); } if (topicRouteData != null) { TopicRouteData old = this.topicRouteTable.get(topic); boolean changed = topicRouteDataIsChange(old, topicRouteData); if (!changed) { changed = this.isNeedUpdateTopicRouteInfo(topic); } else { log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData); } if (changed) { TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData(); for (BrokerData bd : topicRouteData.getBrokerDatas()) { this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs()); } // Update Pub info { TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData); publishInfo.setHaveTopicRouterInfo(true); Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator(); while (it.hasNext()) { Entry<String, MQProducerInner> entry = it.next(); MQProducerInner impl = entry.getValue(); if (impl != null) { impl.updateTopicPublishInfo(topic, publishInfo); } } } // Update sub info { Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData); Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator(); while (it.hasNext()) { Entry<String, MQConsumerInner> entry = it.next(); MQConsumerInner impl = entry.getValue(); if (impl != null) { impl.updateTopicSubscribeInfo(topic, subscribeInfo); } } } log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData); this.topicRouteTable.put(topic, cloneTopicRouteData); return true; } } else { log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}", topic); } } catch (Exception e) { if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC)) { log.warn("updateTopicRouteInfoFromNameServer Exception", e); } } finally { this.lockNamesrv.unlock(); } } else { log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms", LOCK_TIMEOUT_MILLIS); } } catch (InterruptedException e) { log.warn("updateTopicRouteInfoFromNameServer Exception", e); } return false; }
这里首先由mQClientAPIImpl的getTopicRouteInfoFromNameServer方法,从名称服务器上获取其Topic所对应的路由信息
其中Topic的路由信息由TopicRouteData进行封装:
public class TopicRouteData extends RemotingSerializable { private String orderTopicConf; private List<QueueData> queueDatas; private List<BrokerData> brokerDatas; private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable; }
QueueData:
public class QueueData implements Comparable<QueueData> { private String brokerName; private int readQueueNums; private int writeQueueNums; private int perm; private int topicSynFlag; }
BrokerData:
public class BrokerData implements Comparable<BrokerData> { private String cluster; private String brokerName; private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs; }
getTopicRouteInfoFromNameServer方法:
public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis) throws RemotingException, MQClientException, InterruptedException { return getTopicRouteInfoFromNameServer(topic, timeoutMillis, true); } public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis, boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException { GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader(); requestHeader.setTopic(topic); RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINTO_BY_TOPIC, requestHeader); RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis); assert response != null; switch (response.getCode()) { case ResponseCode.TOPIC_NOT_EXIST: { if (allowTopicNotExist && !topic.equals(MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC)) { log.warn("get Topic [{}] RouteInfoFromNameServer is not exist value", topic); } break; } case ResponseCode.SUCCESS: { byte[] body = response.getBody(); if (body != null) { return TopicRouteData.decode(body, TopicRouteData.class); } } default: break; } throw new MQClientException(response.getCode(), response.getRemark()); }
这里主要通过remotingClient即Netty客户端的invokeSync方法向名称服务器发送封装好的request请求来获取response
通过名称服务器寻找与Topic相关的Broker有关路由信息,将这些信息作为response返回,在这里接收到进行处理,封装成TopicRouteData
在invokeSync方法中采用懒加载的方式,尝试获取已经建立好连接的Channel,若是没有,则需要通过bootstrap的connect方法先建立连接产生ChannelFuture,进而获取并缓存Channel
回到updateTopicRouteInfoFromNameServer,通过名称服务器获取到了有关Topic的路由信息,调用topicRouteDataIsChange方法和原来topicRouteTable保存的路由信息进行比较
topicRouteDataIsChange方法:
private boolean topicRouteDataIsChange(TopicRouteData olddata, TopicRouteData nowdata) { if (olddata == null || nowdata == null) return true; TopicRouteData old = olddata.cloneTopicRouteData(); TopicRouteData now = nowdata.cloneTopicRouteData(); Collections.sort(old.getQueueDatas()); Collections.sort(old.getBrokerDatas()); Collections.sort(now.getQueueDatas()); Collections.sort(now.getBrokerDatas()); return !old.equals(now); }
若是没有发生改变,任然要调用isNeedUpdateTopicRouteInfo方法检查是否有需要更新
isNeedUpdateTopicRouteInfo方法:
private boolean isNeedUpdateTopicRouteInfo(final String topic) { boolean result = false; { Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator(); while (it.hasNext() && !result) { Entry<String, MQProducerInner> entry = it.next(); MQProducerInner impl = entry.getValue(); if (impl != null) { result = impl.isPublishTopicNeedUpdate(topic); } } } { Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator(); while (it.hasNext() && !result) { Entry<String, MQConsumerInner> entry = it.next(); MQConsumerInner impl = entry.getValue(); if (impl != null) { result = impl.isSubscribeTopicNeedUpdate(topic); } } } return result; }
分别对所有的消费者和生产者进行检查是否有需要更新有关该Topic的路由信息
当存在需要跟新的情况时,在updateTopicRouteInfoFromNameServer中
首先从topicRouteData中取出BrokerData,即Broker的路由信息,进行更新
再根据topicRouteData从中获取消费者生产者的消息路由信息,分别进行更新
③定时清除离线的Broker,以及向当前在线的Broker发送心跳包
cleanOfflineBroker清除离线的Broker:
private void cleanOfflineBroker() { try { if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) try { ConcurrentHashMap<String, HashMap<Long, String>> updatedTable = new ConcurrentHashMap<String, HashMap<Long, String>>(); Iterator<Entry<String, HashMap<Long, String>>> itBrokerTable = this.brokerAddrTable.entrySet().iterator(); while (itBrokerTable.hasNext()) { Entry<String, HashMap<Long, String>> entry = itBrokerTable.next(); String brokerName = entry.getKey(); HashMap<Long, String> oneTable = entry.getValue(); HashMap<Long, String> cloneAddrTable = new HashMap<Long, String>(); cloneAddrTable.putAll(oneTable); Iterator<Entry<Long, String>> it = cloneAddrTable.entrySet().iterator(); while (it.hasNext()) { Entry<Long, String> ee = it.next(); String addr = ee.getValue(); if (!this.isBrokerAddrExistInTopicRouteTable(addr)) { it.remove(); log.info("the broker addr[{} {}] is offline, remove it", brokerName, addr); } } if (cloneAddrTable.isEmpty()) { itBrokerTable.remove(); log.info("the broker[{}] name's host is offline, remove it", brokerName); } else { updatedTable.put(brokerName, cloneAddrTable); } } if (!updatedTable.isEmpty()) { this.brokerAddrTable.putAll(updatedTable); } } finally { this.lockNamesrv.unlock(); } } catch (InterruptedException e) { log.warn("cleanOfflineBroker Exception", e); } }
这里的brokerAddrTable是会通过②中的定时任务来更新,遍历其中的所有Broker信息,通过isBrokerAddrExistInTopicRouteTable方法,进行检查:
private boolean isBrokerAddrExistInTopicRouteTable(final String addr) { Iterator<Entry<String, TopicRouteData>> it = this.topicRouteTable.entrySet().iterator(); while (it.hasNext()) { Entry<String, TopicRouteData> entry = it.next(); TopicRouteData topicRouteData = entry.getValue(); List<BrokerData> bds = topicRouteData.getBrokerDatas(); for (BrokerData bd : bds) { if (bd.getBrokerAddrs() != null) { boolean exist = bd.getBrokerAddrs().containsValue(addr); if (exist) return true; } } } return false; }
通过比对topicRouteTable中的所有TopicRouteData保存的BrokerAddrs来判断,若是Broker不存在,需要进行清除,进而更新brokerAddrTable
sendHeartbeatToAllBrokerWithLock定时向Broker发送心跳包:
public void sendHeartbeatToAllBrokerWithLock() { if (this.lockHeartbeat.tryLock()) { try { this.sendHeartbeatToAllBroker(); this.uploadFilterClassSource(); } catch (final Exception e) { log.error("sendHeartbeatToAllBroker exception", e); } finally { this.lockHeartbeat.unlock(); } } else { log.warn("lock heartBeat, but failed."); } }
这一部分就不详细介绍,主要还是通过Netty客户端完成心跳包的发送
④定时持久化消费者队列的消费进度,这个在分析消费者时再详细说明
⑤定时调整消费者端的线程池的大小,还是在分析消费者时再详细说明
startScheduledTask创建的五个定时任务结束,回到MQClientInstance的start方法
接着开启pullMessageService服务,为消费者拉取消息
然后开启rebalanceService服务,用来均衡消息队列
这两个服务在有关消费者时再介绍
接着通过:
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
开启push service服务
其中defaultMQProducer是在前面MQClientInstance构造方法中创建的
this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
只不过他调用的start方法,参数为false,也就是没有调用mQClientFactory的start方法
后续会介绍其用途
到这DefaultMQProducerImpl的start方法已经基本完毕,只不过在最后,会通过mQClientFactory的sendHeartbeatToAllBrokerWithLock方法,给所有Broker发送一次心跳包
到此,Producer的启动结束
- Rocketmq之namesrv启动流程源码详解分析
- RocketMQ源码分析之rocketmq-broker启动 (二)
- RocketMQ源码解析-Producer启动
- RocketMQ源码分析(二)Producer端发送数据
- rocketMq-Producer原理源码分析
- RocketMQ源码:Producer启动分析
- RocketMQ-NameServer配置加载、启动源码分析
- Linux内核源码分析--内核启动之(5)Image内核启动(rest_init函数)(Linux-3.0 ARMv7)【转】
- System Server进程启动过程源码分析
- Disconf源码分析之启动过程分析下(2)
- spring启动component-scan类扫描加载过程---源码分析
- S5P210-uboot源码分析-uboot如何启动内核
- 【源码分析】Android系统启动流程.
- Android 源码分析 -- (一) Android启动过程
- Monkey源码分析2—Monkey代码如何被启动执行
- Spark HiveThriftServer2启动流程源码分析
- Activity启动过程源码分析
- 【Spring源码分析系列】启动component-scan类扫描加载过程
- Giraph源码分析(一) —— 启动ZooKeeper服务
- 蔡军生先生第二人生的源码分析(七十四)LLStartUp类维护启动状态机