标签：info upn list rom lte load shm rem server
普通的信息发送和消费
首先要启动nameserver和broker，nameserver是一个几乎无状态节点。broker分为master和slave，master和slave的对应关系通过指定相同的BrokerName，不同的BorkerId来定义，BrokerId为0表示Master，其他为Slave。每个Broker和Name Server集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。
Producer在发送消息前，获取DefaultMQProducer，指定groupName，设置namesrvAddr。开始启动producer，用本身实例defaultMQProducer生成MQClinetManager的实例mQClinetFactory，向mQClinetFactory注册producer也就是向ConcurrentHashMap类型的producerTable中put。然后mQClientFactory.start()。
获取并设置NameServer Addr。
启动通讯服务。
启动定时任务（1：定时获取NameServerAddr，2：定时更新路由信息(也就是更新DefaultMQProducerImpl中的topicPushlishInfoTable)。3：持久化消费记录，4：动态调整线程池）。
pullMessageService.start()。
负载均衡启动
再次调用DefaultMQProducerImp.start(false)这次不走mQClientFactory.start()。
向所有Broker发送心跳，还有一个方法uploadFilterClassSource()(应该是加载过滤器的)
获取完producer(DefaultMQProducer)后，初始化要发送的消息，消息中设置topic，tag，key，消息体(body)。
发送消息。producer.send()
获取路由信息（从topicPublishInfoTable获取topic的TopicPushlishInfo，没有路由信息或 !topicPublishInfo.ok()从NameServer获取，构建请求头，remoteClinic调nameServer获取，看本地是否需要更新，如果需要，更新到本地，获取过来的TopicRouteData转成TopicPublishInfo(producer用的)跟新到本地，也就是向topicPublishInfoTable中put。转成subscribeInfo(consumer用的)向RebalanceImpl.topicSubscribeInfoTable中put）
获取重试次数(timesTotal)，循环这么多次发送消息。获取上一次发送失败的lastBrokerName，第一次发送时，mq为空，如果是轮训，依次轮训topic下的MessageQueue，跳过上次失败的，选取一个mq。如果是顺序消息，就是自己根据算法选择一个MessageQueue列表的中的一个mq。
开始发送。根据上面获取的brokername从MQClientInstance.brokerAddrTable获取brokerAddr，如果为空重复上面的获取路由信息。继续获取。构建SendMessageRequestHeader发送。
=========================================================================================
broker端接受到消息，将消息写入到CommitLog中。
=========================================================================================
Consumer启动前，获取Consumer对象(这里是DefaultMQPushConsumer)，指定groupName，设置NamesrvAddr，指定ConsumeFromWhere，指定订阅topic，push的方法指定MessageListener。consumer.start()启动
复制订阅关系。defaultMQPushConsumer.subscription复制到RebalanceImpl.subscriptionInner。初始化rebalanceImpl对象。构建offsetStore消费对象。consumeMessageService.start()启动消费消息服务。mQClientFactory注册这个消费者(也就是MQClientInstance.consumerTable中put)。mQClientFactory.start()和上面producer启动的过程一样。
在rebalanceService.start()中，进行负载均衡。大体是遍历所有的topic，构建PullRequest，经过多步调用，放在PullMessageService的pullRequestQueue队列中，让pullMessageService线程去获取。
pullMessageService.start()这个是在rebalanceService.start()上面一行启动的。而pullMessageService.start()才是正真的消费。
标签：info upn list rom lte load shm rem server
原文地址：http://www.cnblogs.com/badboyf/p/6114801.html
