public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId) {
    this(clientConfig, instanceIndex, clientId, null);
}

public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
    this.clientConfig = clientConfig;
    this.instanceIndex = instanceIndex;
    // Netty客户端的配置对象
    this.nettyClientConfig = new NettyClientConfig();
    this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
    this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
    // 客户端请求处理器
    this.clientRemotingProcessor = new ClientRemotingProcessor(this);
    // 创建客户端远程通信API实现类的实例
    this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);

    if (this.clientConfig.getNamesrvAddr() != null) {
        // 更新nameserver的地址
        this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
        log.info("user specified name server address: {}", this.clientConfig.getNamesrvAddr());
    }

    // 客户端的id
    this.clientId = clientId;

    // admin控制台的实现
    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());
}

public MQClientAPIImpl(final NettyClientConfig nettyClientConfig,
    final ClientRemotingProcessor clientRemotingProcessor,
    RPCHook rpcHook, final ClientConfig clientConfig) {
    this.clientConfig = clientConfig;
    topAddressing = new TopAddressing(MixAll.getWSAddr(), clientConfig.getUnitName());
    // 创建Netty客户端
    this.remotingClient = new NettyRemotingClient(nettyClientConfig, null);
    this.clientRemotingProcessor = clientRemotingProcessor;

    // 注册RPC回调
    // Inject stream rpc hook first to make reserve field signature
    if (clientConfig.isEnableStreamRequestType()) {
        this.remotingClient.registerRPCHook(new StreamTypeRPCHook());
    }
    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);

    this.remotingClient.registerProcessor(RequestCode.PUSH_REPLY_MESSAGE_TO_CLIENT, this.clientRemotingProcessor, null);
}

public NettyRemotingClient(final NettyClientConfig nettyClientConfig) {
    this(nettyClientConfig, null);
}

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());
        }
    });

    // 创建事件循环组，组内只有1个事件循环
    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()));
        }
    });

}

@Override
public void registerRPCHook(RPCHook rpcHook) {
    // 如果钩子不是null且列表中不存在，才进行添加
    if (rpcHook != null && !rpcHooks.contains(rpcHook)) {
        // 将钩子对象添加到列表中
        rpcHooks.add(rpcHook);
    }
}

@Override
public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
    ExecutorService executorThis = executor;
    if (null == executor) {
        executorThis = this.publicExecutor;
    }

    // 将处理器和线程池绑定在一起
    Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<NettyRequestProcessor, ExecutorService>(processor, executorThis);
    // 将处理器添加到map中
    this.processorTable.put(requestCode, pair);
}

private static MQClientManager instance = new MQClientManager();
private AtomicInteger factoryIndexGenerator = new AtomicInteger();
// 客户端的映射表，键是clientId，值是客户端对象
private ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable =
    new ConcurrentHashMap<String, MQClientInstance>();

public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig) {
    return getOrCreateMQClientInstance(clientConfig, null);
}

public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
    // 生成clientID，组成形式为：IP@instanceName
    String clientId = clientConfig.buildMQClientId();
    // 从客户端的缓存中查找对应id的客户端实例
    MQClientInstance instance = this.factoryTable.get(clientId);
    if (null == instance) {
        // 利用客户端id创建新的客户端对象
        instance =
            new MQClientInstance(clientConfig.cloneClientConfig(),
                this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
        // 加入缓存，这里使用的是ConcurrentHashMap，所以不可能有两个id的客户端对象被添加进入缓存中
        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;
}

// 实例名称
private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");

public String buildMQClientId() {
    StringBuilder sb = new StringBuilder();
    // 拼接IP
    sb.append(this.getClientIP());

    sb.append("@");
    // 拼接实例名称
    sb.append(this.getInstanceName());

    // 默认是null，不满足条件
    if (!UtilAll.isBlank(this.unitName)) {
        sb.append("@");
        // 拼接单元名称
        sb.append(this.unitName);
    }

    // 默认是false
    if (enableStreamRequestType) {
        sb.append("@");
        // 拼接0
        sb.append(RequestType.STREAM);
    }

    return sb.toString();
}

public void changeInstanceNameToPID() {
    if (this.instanceName.equals("DEFAULT")) {
        // 替换instanceName，pid加上时间
        this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
    }
}