1 public class DefaultMQProducer extends ClientConfig implements MQProducer {
 2     protected final transient DefaultMQProducerImpl defaultMQProducerImpl;
 3     
 4     private String createTopicKey = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC = "TBW102";
 5     
 6     private volatile int defaultTopicQueueNums = 4;
 7     
 8     private int sendMsgTimeout = 3000;
 9     
10     private int compressMsgBodyOverHowmuch = 1024 * 4;
11     
12     private int retryTimesWhenSendFailed = 2;
13     
14     private int retryTimesWhenSendAsyncFailed = 2;
15     
16     private boolean retryAnotherBrokerWhenNotStoreOK = false;
17     
18     private int maxMessageSize = 1024 * 1024 * 4; // 4M
19 }

 1 public DefaultMQProducer() {
 2     this(MixAll.DEFAULT_PRODUCER_GROUP, null);
 3 }
 4 
 5 public DefaultMQProducer(final String producerGroup) {
 6     this(producerGroup, null);
 7 }
 8 
 9 public DefaultMQProducer(final String producerGroup, RPCHook rpcHook) {
10     this.producerGroup = producerGroup;
11     defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
12 }

 1 public class ClientConfig {
 2     public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel";
 3     
 4     private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV));
 5     
 6     private String clientIP = RemotingUtil.getLocalAddress();
 7     
 8     private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");
 9     
10     private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();
11     
12     private int pollNameServerInterval = 1000 * 30;
13     
14     private int heartbeatBrokerInterval = 1000 * 30;
15     
16     private boolean vipChannelEnabled = Boolean.parseBoolean(System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "true"));
17 }

 1 private final Random random = new Random();
 2 private final ConcurrentMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable =
 3         new ConcurrentHashMap<String, TopicPublishInfo>();
 4 private final ArrayList<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>();
 5 private final RPCHook rpcHook;
 6 protected BlockingQueue<Runnable> checkRequestQueue;
 7 protected ExecutorService checkExecutor;
 8 private ServiceState serviceState = ServiceState.CREATE_JUST;
 9 private MQClientInstance mQClientFactory;
10 private ArrayList<CheckForbiddenHook> checkForbiddenHookList = new ArrayList<CheckForbiddenHook>();
11 private int zipCompressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5"));
12 private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy();
13 private final BlockingQueue<Runnable> asyncSenderThreadPoolQueue;
14 private final ExecutorService defaultAsyncSenderExecutor;
15 private ExecutorService asyncSenderExecutor;
16 
17 public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) {
18     this.defaultMQProducer = defaultMQProducer;
19     this.rpcHook = rpcHook;
20 
21     this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000);
22     this.defaultAsyncSenderExecutor = new ThreadPoolExecutor(
23         Runtime.getRuntime().availableProcessors(),
24         Runtime.getRuntime().availableProcessors(),
25         1000 * 60,
26         TimeUnit.MILLISECONDS,
27         this.asyncSenderThreadPoolQueue,
28         new ThreadFactory() {
29             private AtomicInteger threadIndex = new AtomicInteger(0);
30 
31             @Override
32             public Thread newThread(Runnable r) {
33                 return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet());
34             }
35         });
36 }

 1 public void start() throws MQClientException {
 2     this.defaultMQProducerImpl.start();
 3     if (null != traceDispatcher) {
 4         try {
 5             traceDispatcher.start(this.getNamesrvAddr());
 6         } catch (MQClientException e) {
 7             log.warn("trace dispatcher start failed ", e);
 8         }
 9     }
10 }

 1 public void start() throws MQClientException {
 2     this.start(true);
 3 }
 4 
 5 public void start(final boolean startFactory) throws MQClientException {
 6     switch (this.serviceState) {
 7         case CREATE_JUST:
 8             this.serviceState = ServiceState.START_FAILED;
 9 
10             this.checkConfig();
11 
12             if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
13                 this.defaultMQProducer.changeInstanceNameToPID();
14             }
15 
16             this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
17 
18             boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
19             if (!registerOK) {
20                 this.serviceState = ServiceState.CREATE_JUST;
21                 throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
22                     + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
23                     null);
24             }
25 
26             this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
27 
28             if (startFactory) {
29                 mQClientFactory.start();
30             }
31 
32             log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
33                 this.defaultMQProducer.isSendMessageWithVIPChannel());
34             this.serviceState = ServiceState.RUNNING;
35             break;
36         case RUNNING:
37         case START_FAILED:
38         case SHUTDOWN_ALREADY:
39             throw new MQClientException("The producer service state not OK, maybe started once, "
40                 + this.serviceState
41                 + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
42                 null);
43         default:
44             break;
45     }
46 
47     this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
48 }

1 public enum ServiceState {
2     CREATE_JUST,
3     RUNNING,
4     SHUTDOWN_ALREADY,
5     START_FAILED;
6 
7     private ServiceState() {
8     }
9 }

 1 this.serviceState = ServiceState.START_FAILED;
 2 
 3 this.checkConfig();
 4 
 5 if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
 6 this.defaultMQProducer.changeInstanceNameToPID();
 7 }
 8 
 9 this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
10 
11 boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
12 if (!registerOK) {
13 this.serviceState = ServiceState.CREATE_JUST;
14 throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
15 + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
16 null);
17 }
18 
19 this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
20 
21 if (startFactory) {
22 mQClientFactory.start();
23 }
24 
25 log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
26 this.defaultMQProducer.isSendMessageWithVIPChannel());
27 this.serviceState = ServiceState.RUNNING;
28 break;

 1 private void checkConfig() throws MQClientException {
 2     Validators.checkGroup(this.defaultMQProducer.getProducerGroup());
 3 
 4     if (null == this.defaultMQProducer.getProducerGroup()) {
 5         throw new MQClientException("producerGroup is null", null);
 6     }
 7 
 8     if (this.defaultMQProducer.getProducerGroup().equals(MixAll.DEFAULT_PRODUCER_GROUP)) {
 9         throw new MQClientException("producerGroup can not equal " + MixAll.DEFAULT_PRODUCER_GROUP + ", please specify another one.",
10             null);
11     }
12 }

1 public static final String DEFAULT_PRODUCER_GROUP = "DEFAULT_PRODUCER";

 1 public class MQClientManager {
 2     private final static InternalLogger log = ClientLogger.getLog();
 3     private static MQClientManager instance = new MQClientManager();
 4     private AtomicInteger factoryIndexGenerator = new AtomicInteger();
 5     private ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable =
 6         new ConcurrentHashMap<String, MQClientInstance>();
 7 
 8     private MQClientManager() {
 9     }
10 
11     public static MQClientManager getInstance() {
12         return instance;
13     }
14 }

 1 public MQClientInstance getAndCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
 2     String clientId = clientConfig.buildMQClientId();
 3     MQClientInstance instance = this.factoryTable.get(clientId);
 4     if (null == instance) {
 5         instance =
 6             new MQClientInstance(clientConfig.cloneClientConfig(),
 7                 this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
 8         MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
 9         if (prev != null) {
10             instance = prev;
11             log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
12         } else {
13             log.info("Created new MQClientInstance for clientId:[{}]", clientId);
14         }
15     }
16 
17     return instance;
18 }

 1 public String buildMQClientId() {
 2     StringBuilder sb = new StringBuilder();
 3     sb.append(this.getClientIP());
 4 
 5     sb.append("@");
 6     sb.append(this.getInstanceName());
 7     if (!UtilAll.isBlank(this.unitName)) {
 8         sb.append("@");
 9         sb.append(this.unitName);
10     }
11 
12     return sb.toString();
13 }

 1 public ClientConfig cloneClientConfig() {
 2     ClientConfig cc = new ClientConfig();
 3     cc.namesrvAddr = namesrvAddr;
 4     cc.clientIP = clientIP;
 5     cc.instanceName = instanceName;
 6     cc.clientCallbackExecutorThreads = clientCallbackExecutorThreads;
 7     cc.pollNameServerInterval = pollNameServerInterval;
 8     cc.heartbeatBrokerInterval = heartbeatBrokerInterval;
 9     cc.persistConsumerOffsetInterval = persistConsumerOffsetInterval;
10     cc.unitMode = unitMode;
11     cc.unitName = unitName;
12     cc.vipChannelEnabled = vipChannelEnabled;
13     cc.useTLS = useTLS;
14     cc.language = language;
15     return cc;
16 }

 1 public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
 2     this.clientConfig = clientConfig;
 3     this.instanceIndex = instanceIndex;
 4     this.nettyClientConfig = new NettyClientConfig();
 5     this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
 6     this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
 7     this.clientRemotingProcessor = new ClientRemotingProcessor(this);
 8     this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);
 9 
10     if (this.clientConfig.getNamesrvAddr() != null) {
11         this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
12         log.info("user specified name server address: {}", this.clientConfig.getNamesrvAddr());
13     }
14 
15     this.clientId = clientId;
16 
17     this.mQAdminImpl = new MQAdminImpl(this);
18 
19     this.pullMessageService = new PullMessageService(this);
20 
21     this.rebalanceService = new RebalanceService(this);
22 
23     this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
24     this.defaultMQProducer.resetClientConfig(clientConfig);
25 
26     this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService);
27 
28     log.info("Created a new client Instance, InstanceIndex:{}, ClientID:{}, ClientConfig:{}, ClientVersion:{}, SerializerType:{}",
29         this.instanceIndex,
30         this.clientId,
31         this.clientConfig,
32         MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION), RemotingCommand.getSerializeTypeConfigInThisServer());
33 }

 1 public MQClientAPIImpl(final NettyClientConfig nettyClientConfig,
 2     final ClientRemotingProcessor clientRemotingProcessor,
 3     RPCHook rpcHook, final ClientConfig clientConfig) {
 4     this.clientConfig = clientConfig;
 5     topAddressing = new TopAddressing(MixAll.getWSAddr(), clientConfig.getUnitName());
 6     this.remotingClient = new NettyRemotingClient(nettyClientConfig, null);
 7     this.clientRemotingProcessor = clientRemotingProcessor;
 8     
 9     this.remotingClient.registerRPCHook(rpcHook);
10     this.remotingClient.registerProcessor(RequestCode.CHECK_TRANSACTION_STATE, this.clientRemotingProcessor, null);
11     
12     this.remotingClient.registerProcessor(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, this.clientRemotingProcessor, null);
13     
14     this.remotingClient.registerProcessor(RequestCode.RESET_CONSUMER_CLIENT_OFFSET, this.clientRemotingProcessor, null);
15     
16     this.remotingClient.registerProcessor(RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT, this.clientRemotingProcessor, null);
17     
18     this.remotingClient.registerProcessor(RequestCode.GET_CONSUMER_RUNNING_INFO, this.clientRemotingProcessor, null);
19     
20     this.remotingClient.registerProcessor(RequestCode.CONSUME_MESSAGE_DIRECTLY, this.clientRemotingProcessor, null);
21 }

1 http://jmenv.tbsite.net:8080/rocketmq/nsaddr

 1 private final Bootstrap bootstrap = new Bootstrap();
 2 // 名称服务列表
 3 private final AtomicReference<List<String>> namesrvAddrList = new AtomicReference<List<String>>();
 4 
 5 public NettyRemotingClient(final NettyClientConfig nettyClientConfig,
 6     final ChannelEventListener channelEventListener) {
 7     super(nettyClientConfig.getClientOnewaySemaphoreValue(), nettyClientConfig.getClientAsyncSemaphoreValue());
 8     this.nettyClientConfig = nettyClientConfig;
 9     this.channelEventListener = channelEventListener;
10 
11     int publicThreadNums = nettyClientConfig.getClientCallbackExecutorThreads();
12     if (publicThreadNums <= 0) {
13         publicThreadNums = 4;
14     }
15 
16     this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
17         private AtomicInteger threadIndex = new AtomicInteger(0);
18 
19         @Override
20         public Thread newThread(Runnable r) {
21             return new Thread(r, "NettyClientPublicExecutor_" + this.threadIndex.incrementAndGet());
22         }
23     });
24 
25     this.eventLoopGroupWorker = new NioEventLoopGroup(1, new ThreadFactory() {
26         private AtomicInteger threadIndex = new AtomicInteger(0);
27 
28         @Override
29         public Thread newThread(Runnable r) {
30             return new Thread(r, String.format("NettyClientSelector_%d", this.threadIndex.incrementAndGet()));
31         }
32     });
33 
34     if (nettyClientConfig.isUseTLS()) {
35         try {
36             sslContext = TlsHelper.buildSslContext(true);
37             log.info("SSL enabled for client");
38         } catch (IOException e) {
39             log.error("Failed to create SSLContext", e);
40         } catch (CertificateException e) {
41             log.error("Failed to create SSLContext", e);
42             throw new RuntimeException("Failed to create SSLContext", e);
43         }
44     }
45 }

1 public void updateNameServerAddressList(final String addrs) {
2     String[] addrArray = addrs.split(";");
3     List<String> list = Arrays.asList(addrArray);
4     this.remotingClient.updateNameServerAddressList(list);
5 }

 1 public void updateNameServerAddressList(List<String> addrs) {
 2     List<String> old = this.namesrvAddrList.get();
 3     boolean update = false;
 4 
 5     if (!addrs.isEmpty()) {
 6         if (null == old) {
 7             update = true;
 8         } else if (addrs.size() != old.size()) {
 9             update = true;
10         } else {
11             for (int i = 0; i < addrs.size() && !update; i++) {
12                 if (!old.contains(addrs.get(i))) {
13                     update = true;
14                 }
15             }
16         }
17 
18         if (update) {
19             Collections.shuffle(addrs);
20             log.info("name server address updated. NEW : {} , OLD: {}", addrs, old);
21             this.namesrvAddrList.set(addrs);
22         }
23     }
24 }

 1 public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
 2     if (null == group || null == producer) {
 3         return false;
 4     }
 5 
 6     MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
 7     if (prev != null) {
 8         log.warn("the producer group[{}] exist already.", group);
 9         return false;
10     }
11 
12     return true;
13 }

1 private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
2 private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
3 private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
4 private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
5         new ConcurrentHashMap<String, HashMap<Long, String>>();

1 public class TopicPublishInfo {
2     private boolean orderTopic = false;
3     private boolean haveTopicRouterInfo = false;
4     private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>();
5     private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
6 }

 1 public class MessageQueue implements Comparable<MessageQueue>, Serializable {
 2     private static final long serialVersionUID = 6191200464116433425L;
 3     private String topic;
 4     private String brokerName;
 5     private int queueId;
 6 
 7     public MessageQueue() {
 8     }
 9 
10     public MessageQueue(String topic, String brokerName, int queueId) {
11         this.topic = topic;
12         this.brokerName = brokerName;
13         this.queueId = queueId;
14     }
15 }

 1 public class ThreadLocalIndex {
 2     private final ThreadLocal<Integer> threadLocalIndex = new ThreadLocal<Integer>();
 3     private final Random random = new Random();
 4 
 5     public int getAndIncrement() {
 6         Integer index = this.threadLocalIndex.get();
 7         if (null == index) {
 8             index = Math.abs(random.nextInt());
 9             if (index < 0)
10                 index = 0;
11             this.threadLocalIndex.set(index);
12         }
13 
14         index = Math.abs(index + 1);
15         if (index < 0)
16             index = 0;
17 
18         this.threadLocalIndex.set(index);
19         return index;
20     }
21 
22     @Override
23     public String toString() {
24         return "ThreadLocalIndex{" +
25             "threadLocalIndex=" + threadLocalIndex.get() +
26             \'}\';
27     }
28 }

 1 public void start() throws MQClientException {
 2     synchronized (this) {
 3         switch (this.serviceState) {
 4             case CREATE_JUST:
 5                 this.serviceState = ServiceState.START_FAILED;
 6                 // If not specified,looking address from name server
 7                 if (null == this.clientConfig.getNamesrvAddr()) {
 8                     this.mQClientAPIImpl.fetchNameServerAddr();
 9                 }
10                 // Start request-response channel
11                 this.mQClientAPIImpl.start();
12                 // Start various schedule tasks
13                 this.startScheduledTask();
14                 // Start pull service
15                 this.pullMessageService.start();
16                 // Start rebalance service
17                 this.rebalanceService.start();
18                 // Start push service
19                 this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
20                 log.info("the client factory [{}] start OK", this.clientId);
21                 this.serviceState = ServiceState.RUNNING;
22                 break;
23             case RUNNING:
24                 break;
25             case SHUTDOWN_ALREADY:
26                 break;
27             case START_FAILED:
28                 throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
29             default:
30                 break;
31         }
32     }
33 }

 1 public String fetchNameServerAddr() {
 2     try {
 3         String addrs = this.topAddressing.fetchNSAddr();
 4         if (addrs != null) {
 5             if (!addrs.equals(this.nameSrvAddr)) {
 6                 log.info("name server address changed, old=" + this.nameSrvAddr + ", new=" + addrs);
 7                 this.updateNameServerAddressList(addrs);
 8                 this.nameSrvAddr = addrs;
 9                 return nameSrvAddr;
10             }
11         }
12     } catch (Exception e) {
13         log.error("fetchNameServerAddr Exception", e);
14     }
15     return nameSrvAddr;
16 }

 1 public final String fetchNSAddr() {
 2     return fetchNSAddr(true, 3000);
 3 }
 4 
 5 public final String fetchNSAddr(boolean verbose, long timeoutMills) {
 6     String url = this.wsAddr;
 7     try {
 8         if (!UtilAll.isBlank(this.unitName)) {
 9             url = url + "-" + this.unitName + "?nofix=1";
10         }
11         HttpTinyClient.HttpResult result = HttpTinyClient.httpGet(url, null, null, "UTF-8", timeoutMills);
12         if (200 == result.code) {
13             String responseStr = result.content;
14             if (responseStr != null) {
15                 return clearNewLine(responseStr);
16             } else {
17                 log.error("fetch nameserver address is null");
18             }
19         } else {
20             log.error("fetch nameserver address failed. statusCode=" + result.code);
21         }
22     } catch (IOException e) {
23         if (verbose) {
24             log.error("fetch name server address exception", e);
25         }
26     }
27 
28     if (verbose) {
29         String errorMsg =
30             "connect to " + url + " failed, maybe the domain name " + MixAll.getWSAddr() + " not bind in /etc/hosts";
31         errorMsg += FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL);
32 
33         log.warn(errorMsg);
34     }
35     return null;
36 }

 1 static public HttpResult httpGet(String url, List<String> headers, List<String> paramValues,
 2     String encoding, long readTimeoutMs) throws IOException {
 3     String encodedContent = encodingParams(paramValues, encoding);
 4     url += (null == encodedContent) ? "" : ("?" + encodedContent);
 5 
 6     HttpURLConnection conn = null;
 7     try {
 8         conn = (HttpURLConnection) new URL(url).openConnection();
 9         conn.setRequestMethod("GET");
10         conn.setConnectTimeout((int) readTimeoutMs);
11         conn.setReadTimeout((int) readTimeoutMs);
12         setHeaders(conn, headers, encoding);
13 
14         conn.connect();
15         int respCode = conn.getResponseCode();
16         String resp = null;
17 
18         if (HttpURLConnection.HTTP_OK == respCode) {
19             resp = IOTinyUtils.toString(conn.getInputStream(), encoding);
20         } else {
21             resp = IOTinyUtils.toString(conn.getErrorStream(), encoding);
22         }
23         return new HttpResult(respCode, resp);
24     } finally {
25         if (conn != null) {
26             conn.disconnect();
27         }
28     }
29 }

1 public void start() {
2     this.remotingClient.start();
3 }

 1 public void start() {
 2     this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
 3         nettyClientConfig.getClientWorkerThreads(),
 4         new ThreadFactory() {
 5 
 6             private AtomicInteger threadIndex = new AtomicInteger(0);
 7 
 8             @Override
 9             public Thread newThread(Runnable r) {
10                 return new Thread(r, "NettyClientWorkerThread_" + this.threadIndex.incrementAndGet());
11             }
12         });
13 
14     Bootstrap handler = this.bootstrap.group(this.eventLoopGroupWorker).channel(NioSocketChannel.class)
15         .option(ChannelOption.TCP_NODELAY, true)
16         .option(ChannelOption.SO_KEEPALIVE, false)
17         .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis())
18         .option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize())
19         .option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize())
20         .handler(new ChannelInitializer<SocketChannel>() {
21             @Override
22             public void initChannel(SocketChannel ch) throws Exception {
23                 ChannelPipeline pipeline = ch.pipeline();
24                 if (nettyClientConfig.isUseTLS()) {
25                     if (null != sslContext) {
26                         pipeline.addFirst(defaultEventExecutorGroup, "sslHandler", sslContext.newHandler(ch.alloc()));
27                         log.info("Prepend SSL handler");
28                     } else {
29                         log.warn("Connections are insecure as SSLContext is null!");
30                     }
31                 }
32                 pipeline.addLast(
33                     defaultEventExecutorGroup,
34                     new NettyEncoder(),
35                     new NettyDecoder(),
36                     new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),
37                     new NettyConnectManageHandler(),
38                     new NettyClientHandler());
39             }
40         });
41 
42     this.timer.scheduleAtFixedRate(new TimerTask() {
43         @Override
44         public void run() {
45             try {
46                 NettyRemotingClient.this.scanResponseTable();
47             } catch (Throwable e) {
48                 log.error("scanResponseTable exception", e);
49             }
50         }
51     }, 1000 * 3, 1000);
52 
53     if (this.channelEventListener != null) {
54         this.nettyEventExecutor.start();
55     }
56 }