public Consumer(string id, string groupName, ConsumerSetting setting)
        {
            if (id == null)
            {
                throw new ArgumentNullException("id");
            }
            if (groupName == null)
            {
                throw new ArgumentNullException("groupName");
            }
            Id = id;
            GroupName = groupName;
            Setting = setting ?? new ConsumerSetting();

            _lockObject = new object();
            _subscriptionTopics = new List<string>();
            _topicQueuesDict = new ConcurrentDictionary<string, IList<MessageQueue>>();
            _pullRequestQueue = new BlockingCollection<PullRequest>(new ConcurrentQueue<PullRequest>());
            _pullRequestDict = new ConcurrentDictionary<string, PullRequest>();
            _consumingMessageQueue = new BlockingCollection<ConsumingMessage>(new ConcurrentQueue<ConsumingMessage>());
            _messageRetryQueue = new BlockingCollection<ConsumingMessage>(new ConcurrentQueue<ConsumingMessage>());
            _handlingMessageDict = new ConcurrentDictionary<long, ConsumingMessage>();
            _taskIds = new List<int>();
            _remotingClient = new SocketRemotingClient(Setting.BrokerAddress, Setting.BrokerPort);
            _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
            _scheduleService = ObjectContainer.Resolve<IScheduleService>();
            _allocateMessageQueueStragegy = ObjectContainer.Resolve<IAllocateMessageQueueStrategy>();
            _executePullRequestWorker = new Worker("Consumer.ExecutePullRequest", ExecutePullRequest);
            _handleMessageWorker = new Worker("Consumer.HandleMessage", HandleMessage);
            _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);
        }

 public MessageService(IBinarySerializer binarySerializer)
 {
     _remotingClient = new SocketRemotingClient(Settings.BrokerAddress, Settings.BrokerPort);
     _remotingClient.Connect();
     _remotingClient.Start();
     _binarySerializer = binarySerializer;
 }

        public ClientService(ClientSetting setting, Producer producer, Consumer consumer)
        {
            Ensure.NotNull(setting, "setting");
            if (producer == null && consumer == null)
            {
                throw new ArgumentException("producer or consumer must set at least one of them.");
            }
            else if (producer != null && consumer != null)
            {
                throw new ArgumentException("producer or consumer cannot set both of them.");
            }

            Interlocked.Increment(ref _instanceNumber);

            _producer = producer;
            _consumer = consumer;
            _setting = setting;
            _clientId = BuildClientId(setting.ClientName);
            _brokerConnectionDict = new ConcurrentDictionary<string, BrokerConnection>();
            _topicMessageQueueDict = new ConcurrentDictionary<string, IList<MessageQueue>>();
            _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
            _jsonSerializer = ObjectContainer.Resolve<IJsonSerializer>();
            _scheduleService = ObjectContainer.Resolve<IScheduleService>();
            _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);
            _nameServerRemotingClientList = RemotingClientUtils.CreateRemotingClientList(_setting.NameServerList, _setting.SocketSetting).ToList();
        }

 public QueryTopicConsumeInfoRequestHandler()
 {
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _offsetManager = ObjectContainer.Resolve<IOffsetManager>();
     _queueService = ObjectContainer.Resolve<IQueueService>();
     _consumerManager = ObjectContainer.Resolve<ConsumerManager>();
 }

 public QueryConsumerInfoRequestHandler()
 {
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _offsetStore = ObjectContainer.Resolve<IConsumeOffsetStore>();
     _consumerManager = ObjectContainer.Resolve<ConsumerManager>();
     _queueService = ObjectContainer.Resolve<IQueueStore>();
 }

        public Consumer(string groupName, ConsumerSetting setting, string consumerName = null)
        {
            if (groupName == null)
            {
                throw new ArgumentNullException("groupName");
            }

            Name = consumerName;
            GroupName = groupName;
            Setting = setting ?? new ConsumerSetting();

            if (Setting.NameServerList == null || Setting.NameServerList.Count() == 0)
            {
                throw new Exception("Name server address is not specified.");
            }

            _subscriptionTopics = new Dictionary<string, HashSet<string>>();
            _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
            _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);

            var clientSetting = new ClientSetting
            {
                ClientName = Name,
                ClusterName = Setting.ClusterName,
                NameServerList = Setting.NameServerList,
                SocketSetting = Setting.SocketSetting,
                OnlyFindMasterBroker = true,
                SendHeartbeatInterval = Setting.HeartbeatBrokerInterval,
                RefreshBrokerAndTopicRouteInfoInterval = Setting.RefreshBrokerAndTopicRouteInfoInterval
            };
            _clientService = new ClientService(clientSetting, null, this);
            _pullMessageService = new PullMessageService(this, _clientService);
            _commitConsumeOffsetService = new CommitConsumeOffsetService(this, _clientService);
            _rebalanceService = new RebalanceService(this, _clientService, _pullMessageService, _commitConsumeOffsetService);
        }

        public Consumer(string groupName, ConsumerSetting setting)
        {
            if (groupName == null)
            {
                throw new ArgumentNullException("groupName");
            }
            GroupName = groupName;
            Setting = setting ?? new ConsumerSetting();

            _lockObject = new object();
            _subscriptionTopics = new Dictionary<string, HashSet<string>>();
            _topicQueuesDict = new ConcurrentDictionary<string, IList<MessageQueue>>();
            _pullRequestDict = new ConcurrentDictionary<string, PullRequest>();
            _remotingClient = new SocketRemotingClient(Setting.BrokerAddress, Setting.SocketSetting, Setting.LocalAddress);
            _adminRemotingClient = new SocketRemotingClient(Setting.BrokerAdminAddress, Setting.SocketSetting, Setting.LocalAdminAddress);
            _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
            _scheduleService = ObjectContainer.Resolve<IScheduleService>();
            _allocateMessageQueueStragegy = ObjectContainer.Resolve<IAllocateMessageQueueStrategy>();
            _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);

            _remotingClient.RegisterConnectionEventListener(new ConnectionEventListener(this));

            if (Setting.MessageHandleMode == MessageHandleMode.Sequential)
            {
                _consumingMessageQueue = new BlockingCollection<ConsumingMessage>();
                _consumeMessageWorker = new Worker("ConsumeMessage", () => HandleMessage(_consumingMessageQueue.Take()));
            }
            _messageRetryQueue = new BlockingCollection<ConsumingMessage>();
        }

 public SendMessageRequestHandler(BrokerController brokerController)
 {
     _brokerController = brokerController;
     _messageService = ObjectContainer.Resolve<IMessageService>();
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().Name);
 }

        public Consumer(string groupName, ConsumerSetting setting)
        {
            if (groupName == null)
            {
                throw new ArgumentNullException("groupName");
            }
            GroupName = groupName;
            Setting = setting ?? new ConsumerSetting();

            _lockObject = new object();
            _subscriptionTopics = new Dictionary<string, HashSet<string>>();
            _topicQueuesDict = new ConcurrentDictionary<string, IList<MessageQueue>>();
            _pullRequestQueue = new BlockingCollection<PullRequest>(new ConcurrentQueue<PullRequest>());
            _pullRequestDict = new ConcurrentDictionary<string, PullRequest>();
            _messageRetryQueue = new BlockingCollection<ConsumingMessage>(new ConcurrentQueue<ConsumingMessage>());
            _taskFactory = new TaskFactory(new LimitedConcurrencyLevelTaskScheduler(Setting.ConsumeThreadMaxCount));
            _remotingClient = new SocketRemotingClient(Setting.BrokerAddress, Setting.SocketSetting, Setting.LocalAddress);
            _adminRemotingClient = new SocketRemotingClient(Setting.BrokerAdminAddress, Setting.SocketSetting, Setting.LocalAdminAddress);
            _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
            _scheduleService = ObjectContainer.Resolve<IScheduleService>();
            _allocateMessageQueueStragegy = ObjectContainer.Resolve<IAllocateMessageQueueStrategy>();
            _executePullRequestWorker = new Worker("ExecutePullRequest", ExecutePullRequest);
            _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);

            _remotingClient.RegisterConnectionEventListener(new ConnectionEventListener(this));
        }

        EventStoreHolder(IHostConfiguration configuration, IBinarySerializer serializer)
        {
            var ipEndpoint = new IPEndPoint(configuration.EventStoreIp, configuration.EventStorePort);

            _connection = EventStoreConnection.Create(ipEndpoint);
            _serializer = serializer;
        }

 public MessageService(IBinarySerializer binarySerializer, IScheduleService scheduleService, SendEmailService sendEmailService)
 {
     _nameServerRemotingClientList = CreateRemotingClientList(Settings.NameServerList);
     _clusterBrokerDict = new ConcurrentDictionary<string, IList<BrokerClient>>();
     _binarySerializer = binarySerializer;
     _scheduleService = scheduleService;
     _sendEmailService = sendEmailService;
 }

 public EventPublisher(ProducerSetting setting, string id)
 {
     _producer = new Producer(setting, id);
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _eventTopicProvider = ObjectContainer.Resolve<IEventTopicProvider>();
     _eventTypeCodeProvider = ObjectContainer.Resolve<IEventTypeCodeProvider>();
     _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().Name);
 }

 /// <summary>Parameterized constructor.
 /// </summary>
 public SqlServerEventStore(string connectionString, string eventTable, string commitIndexName, string versionIndexName)
 {
     _connectionString = connectionString;
     _eventTable = eventTable;
     _commitIndexName = commitIndexName;
     _versionIndexName = versionIndexName;
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _connectionFactory = ObjectContainer.Resolve<IDbConnectionFactory>();
 }

 public ClusterManager(NameServerController nameServerController)
 {
     _scheduleService = ObjectContainer.Resolve<IScheduleService>();
     _jsonSerializer = ObjectContainer.Resolve<IJsonSerializer>();
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _clusterDict = new ConcurrentDictionary<string, Cluster>();
     _nameServerController = nameServerController;
     _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);
 }

 public EventConsumer(string id, ConsumerSetting setting, string groupName, DomainEventHandledMessageSender domainEventHandledMessageSender)
 {
     _consumer = new Consumer(id, setting, string.IsNullOrEmpty(groupName) ? typeof(EventConsumer).Name + "Group" : groupName);
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _eventTypeCodeProvider = ObjectContainer.Resolve<IEventTypeCodeProvider>();
     _eventProcessor = ObjectContainer.Resolve<IEventProcessor>();
     _messageContextDict = new ConcurrentDictionary<string, IMessageContext>();
     _domainEventHandledMessageSender = domainEventHandledMessageSender;
 }

 public MessageService(IBinarySerializer binarySerializer, IScheduleService scheduleService, SendEmailService sendEmailService)
 {
     _remotingClient = new SocketRemotingClient(Settings.BrokerAddress);
     _binarySerializer = binarySerializer;
     _scheduleService = scheduleService;
     _unconsumedMessageWarnningThreshold = int.Parse(ConfigurationManager.AppSettings["unconsumedMessageWarnningThreshold"]);
     _checkUnconsumedMessageInterval = int.Parse(ConfigurationManager.AppSettings["checkUnconsumedMessageInterval"]);
     _sendEmailService = sendEmailService;
 }

 public CommandService(CommandResultProcessor commandResultProcessor, ProducerSetting setting, string id)
 {
     _commandResultProcessor = commandResultProcessor;
     _producer = new Producer(setting, id);
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _commandTopicProvider = ObjectContainer.Resolve<ICommandTopicProvider>();
     _commandTypeCodeProvider = ObjectContainer.Resolve<ICommandTypeCodeProvider>();
     _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().Name);
 }

 /// <summary>
 /// Ctor
 /// </summary>
 /// <param name="serviceEvents">ServiceEvents</param>
 /// <param name="listenerFactory">KafkaListener factory</param>
 /// <param name="serializer">Binary serializer</param>
 public KafkaHostListener(
     IServiceEvents serviceEvents,
     IKafkaListenerFactory listenerFactory,
     IBinarySerializer serializer)
 {
     _serviceEvents = serviceEvents;
     _listenerFactory = listenerFactory;
     _serializer = serializer;
 }

 public CommitConsumeOffsetService(Consumer consumer, ClientService clientService)
 {
     _consumeOffsetInfoDict = new ConcurrentDictionary<string, ConsumeOffsetInfo>();
     _consumer = consumer;
     _clientService = clientService;
     _clientId = clientService.GetClientId();
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _scheduleService = ObjectContainer.Resolve<IScheduleService>();
     _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);
 }

 public PullMessageRequestHandler()
 {
     _consumerManager = ObjectContainer.Resolve<ConsumerManager>();
     _suspendedPullRequestManager = ObjectContainer.Resolve<SuspendedPullRequestManager>();
     _messageService = ObjectContainer.Resolve<IMessageService>();
     _queueService = ObjectContainer.Resolve<IQueueService>();
     _offsetManager = ObjectContainer.Resolve<IOffsetManager>();
     _binarySerializer = ObjectContainer.Resolve<IBinarySerializer>();
     _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);
     EmptyResponseData = new byte[0];
 }