RocketMQ - 生产者启动流程

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);
    }
    if (enableStreamRequestType) {
        sb.append("@");
        sb.append(RequestType.STREAM);
    }
    return sb.toString();
}

public class MQClientInstance { 
    ...
    private final ClientConfig clientConfig;
    private final int instanceIndex;
    private final String clientId;
    private final long bootTimestamp = System.currentTimeMillis();
    //当前client实例的全部生产者的内部实例。
    private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
    //当前client实例的全部消费者的内部实例。
    private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
    //当前client实例的全部管理实例。
    private final ConcurrentMap<String/* group */, MQAdminExtInner> adminExtTable = new ConcurrentHashMap<String, MQAdminExtInner>();
    private final NettyClientConfig nettyClientConfig;
    //其实每个client也是一个Netty Server，也会支持Broker访问，这里实现了全部client支持的接口
    private final MQClientAPIImpl mQClientAPIImpl;
    //管理接口的本地实现类
    private final MQAdminImpl mQAdminImpl;
    //当前生产者、消费者中全部Topic的本地缓存路由信息
    private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
    private final Lock lockNamesrv = new ReentrantLock();
    private final Lock lockHeartbeat = new ReentrantLock();
    private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
        new ConcurrentHashMap<String, HashMap<Long, String>>();
    private final ConcurrentMap<String/* Broker Name */, HashMap<String/* address */, Integer>> brokerVersionTable =
        new ConcurrentHashMap<String, HashMap<String, Integer>>();
    //本地定时任务，比如定期获取当前 Namesrv 地址、定期同步Namesrv信息、定期更新Topic路由信息、定期发送心跳信息给Broker、定期清理已下线的Broker、
    //定期持久化消费位点、定期调整消费线程数（这部分源代码被官方删除了）
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "MQClientFactoryScheduledThread");
        }
    });
    //请求的处理器，从处理方法 processRequest() 中我们可以知道目前支持哪些功能接口
    private final ClientRemotingProcessor clientRemotingProcessor;
    //Pull服务
    private final PullMessageService pullMessageService;
    //重新平衡服务。定期执行重新平衡方法this.mqClientFactory.doRebalance（）。
    //这里的 mqClientFactory 就是 MQClientInstance 实例，通过依次调用MQClientInstance中保存
    //的消费者实例的doRebalance（）方法，来感知订阅关系的变化、集群变化等，以达到重新平衡。
    private final RebalanceService rebalanceService;
    private final DefaultMQProducer defaultMQProducer;
    //消费监控 。 比如拉取 RT（Response Time，响应时间）、拉取TPS（Transactions Per Second，每秒处理消息数）、消费RT等都可以统计。
    private final ConsumerStatsManager consumerStatsManager;
 }

public class MQClientInstance {  
    //从多个Namesrv中获取最新Topic路由信息，更新本地缓存
    public void updateTopicRouteInfoFromNameServer(){}
    //清理已经下线的Broker
    private void cleanOfflineBroker(){}
    //检查Client是否在Broker中有效
    private void checkClientInBroker(){}
    //发送客户端的心跳信息给所有的Broker。
    public void sendHeartbeatToAllBrokerWithLock() {}
    //在本地注册一个消费者。
    public synchronized boolean registerConsumer(final String group, final MQConsumerInner consumer) {}
    //取消本地注册的消费者。
    public synchronized void unregisterConsumer(final String group) {}
    //在本地注册一个生产者。
    public synchronized boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {}
    //取消本地注册的生产者。
    public synchronized void unregisterProducer(final String group){}
    //注册一个管理实例。
    public boolean registerAdminExt(final String group, final MQAdminExtInner admin)
    //立即执行一次 Rebalance。该操作是通过 RocketMQ 的一个CountDownLatch2锁来实现的。
    public void rebalanceImmediately(){}
    //对于所有已经注册的消费者实例 ，执行一次Rebalance。
    public void doRebalance(){}
    //在本地缓存中查找Slave Broker信息
    public FindBrokerResult findBrokerAddressInSubscribe(
        final String brokerName,
        final long brokerId,
        final boolean onlyThisBroker
    ){}
    //在本地缓存中查找Master Broker地址。
    public String findBrokerAddressInPublish(final String brokerName){}
    //查找消费者id列表。
    public List<String> findConsumerIdList(final String topic, final String group) {}
    //通过Topic名字查找Broker地址。
    public String findBrokerAddrByTopic(final String topic) {}
    //重置消费位点。
    public synchronized void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable){}
    //获取一个订阅关系中每个队列的消费进度。
    public Map<MessageQueue, Long> getConsumerStatus(String topic, String group){}
    //获取本地缓存Topic路由。
    public ConcurrentMap<String, TopicRouteData> getTopicRouteTable(){}
    //直接将消息发送给指定的消费者消费，和正常投递不同的是，指定了已经订阅的消费者组中的一个，
    //而不是全部已经订阅的消费者。一般适用于在消费消息后，某一个消费者组想再消费一次的场景。    
    public ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg,
        final String consumerGroup,
        final String brokerName) {}
    //获取消费者的消费统计信息。包含消费RT、消费TPS等。
    public ConsumerRunningInfo consumerRunningInfo(final String consumerGroup){}
 }