public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
 {
     var current = Interlocked.Increment(ref totalHandled);
     if (current == 1)
     {
         watch = Stopwatch.StartNew();
     }
     if (storeOption == "UnManagedMemory")
     {
         SaveMessage(remotingRequest.Body);
     }
     else if (storeOption == "ManagedMemory")
     {
         messageDictionary[remotingRequest.Sequence] = remotingRequest.Body;
     }
     else if (storeOption == "OnlyMapping")
     {
         messageDictionary[remotingRequest.Sequence] = remotingRequest.Sequence;
     }
     if (current % 10000 == 0)
     {
         Console.WriteLine("Handled request, size:{0}, count:{1}, timeSpent: {2}ms", remotingRequest.Body.Length, current, watch.ElapsedMilliseconds);
     }
     return new RemotingResponse(remotingRequest.Code, 10, remotingRequest.Type, response, remotingRequest.Sequence);
 }

        public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest request)
        {
            var queryTopicConsumeInfoRequest = _binarySerializer.Deserialize<QueryTopicConsumeInfoRequest>(request.Body);
            var topicConsumeInfoList = new List<TopicConsumeInfo>();

            if (!string.IsNullOrEmpty(queryTopicConsumeInfoRequest.GroupName))
            {
                var consumerGroups = _brokerController.ConsumerManager.QueryConsumerGroup(queryTopicConsumeInfoRequest.GroupName);
                foreach (var consumerGroup in consumerGroups)
                {
                    foreach (var topicConsumeInfo in GetTopicConsumeInfoForGroup(consumerGroup, queryTopicConsumeInfoRequest.Topic))
                    {
                        topicConsumeInfoList.Add(topicConsumeInfo);
                    }
                }
            }
            else
            {
                var consumerGroups = _brokerController.ConsumerManager.GetAllConsumerGroups();
                foreach (var consumerGroup in consumerGroups)
                {
                    foreach (var topicConsumeInfo in GetTopicConsumeInfoForGroup(consumerGroup, queryTopicConsumeInfoRequest.Topic))
                    {
                        topicConsumeInfoList.Add(topicConsumeInfo);
                    }
                }
            }

            var data = _binarySerializer.Serialize(topicConsumeInfoList);
            return new RemotingResponse((int)ResponseCode.Success, request.Sequence, data);
        }

        public void SendReplyAsync(int replyType, object replyData, string replyAddress)
        {
            _taskFactory.StartNew(obj =>
            {
                var request = obj as SendReplyRequest;
                try
                {
                    var remotingClientWrapper = GetRemotingClientWrapper(request.ReplyAddress);
                    if (remotingClientWrapper == null) return;

                    var message = _jsonSerializer.Serialize(request.ReplyData);
                    var body = Encoding.UTF8.GetBytes(message);
                    var remotingRequest = new RemotingRequest(request.ReplyType, body);
                    var getContextInfo = new Func<String>(() => string.Format("[replyAddress: {0}, replyType: {1}, message: {2}]", request.ReplyAddress, request.ReplyType, message));

                    _ioHelper.TryIOAction("SendCommandReply", getContextInfo, () =>
                    {
                        _ioHelper.TryIOAction(() =>
                        {
                            var remotingResponse = remotingClientWrapper.RemotingClient.InvokeSync(remotingRequest);
                            if (remotingResponse.Code != Constants.SuccessResponseCode)
                            {
                                throw new IOException("Send command reply failed, remotingResponseCode: {0}", remotingResponse.Code);
                            }
                            remotingClientWrapper.LastActiveTime = DateTime.Now;
                        }, "SendCommandReply");
                    }, 3);
                }
                catch (Exception ex)
                {
                    _logger.Error("Send command reply failed.", ex);
                }
            }, new SendReplyRequest(replyType, replyData, replyAddress));
        }

 public ResponseFuture(RemotingRequest request, long timeoutMillis, TaskCompletionSource<RemotingResponse> taskSource)
 {
     Request = request;
     TimeoutMillis = timeoutMillis;
     _taskSource = taskSource;
     BeginTime = DateTime.Now;
 }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
 {
     var topic = Encoding.UTF8.GetString(remotingRequest.Body);
     var queueIds = _queueService.GetQueues(topic).Select(x => x.QueueId).ToList();
     var data = Encoding.UTF8.GetBytes(string.Join(",", queueIds));
     return RemotingResponseFactory.CreateResponse(remotingRequest, data);
 }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
 {
     var request = _binarySerializer.Deserialize<GetTopicAccumulateInfoListRequest>(remotingRequest.Body);
     var topicAccumulateInfoList = _clusterManager.GetTopicAccumulateInfoList(request);
     var data = _binarySerializer.Serialize(topicAccumulateInfoList);
     return RemotingResponseFactory.CreateResponse(remotingRequest, data);
 }

        public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
        {
            var request = _binarySerializer.Deserialize<QueryTopicQueueInfoRequest>(remotingRequest.Body);
            var topicQueueInfoList = new List<TopicQueueInfo>();
            var topicList = !string.IsNullOrEmpty(request.Topic) ? new List<string> { request.Topic } : _queueService.GetAllTopics().ToList();

            foreach (var topic in topicList)
            {
                var queues = _queueService.QueryQueues(topic).ToList();
                foreach (var queue in queues)
                {
                    var topicQueueInfo = new TopicQueueInfo();
                    topicQueueInfo.Topic = queue.Topic;
                    topicQueueInfo.QueueId = queue.QueueId;
                    topicQueueInfo.QueueCurrentOffset = queue.CurrentOffset;
                    topicQueueInfo.QueueMinOffset = queue.GetMinQueueOffset();
                    topicQueueInfo.QueueMessageCount = queue.GetMessageRealCount();
                    topicQueueInfo.QueueMaxConsumedOffset = _offsetManager.GetMinOffset(queue.Topic, queue.QueueId);
                    topicQueueInfo.Status = queue.Status;
                    topicQueueInfoList.Add(topicQueueInfo);
                }
            }

            return RemotingResponseFactory.CreateResponse(remotingRequest, _binarySerializer.Serialize(topicQueueInfoList));
        }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
 {
     var request = _binarySerializer.Deserialize<GetClusterBrokersRequest>(remotingRequest.Body);
     var brokerInfoList = _clusterManager.GetClusterBrokers(request);
     var data = _binarySerializer.Serialize(brokerInfoList);
     return RemotingResponseFactory.CreateResponse(remotingRequest, data);
 }

        public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest request)
        {
            var queryTopicConsumeInfoRequest = _binarySerializer.Deserialize<QueryTopicQueueInfoRequest>(request.Body);
            var topicQueueInfoList = new List<TopicQueueInfo>();
            var topicList = !string.IsNullOrEmpty(queryTopicConsumeInfoRequest.Topic) ? new List<string> { queryTopicConsumeInfoRequest.Topic } : _messageService.GetAllTopics().ToList();

            foreach (var topic in topicList)
            {
                var queues = _messageService.QueryQueues(topic).ToList();
                foreach (var queue in queues)
                {
                    var queueMinOffset = queue.GetMinQueueOffset();
                    var topicQueueInfo = new TopicQueueInfo();
                    topicQueueInfo.Topic = queue.Topic;
                    topicQueueInfo.QueueId = queue.QueueId;
                    topicQueueInfo.QueueCurrentOffset = queue.CurrentOffset;
                    topicQueueInfo.QueueMinOffset = queueMinOffset != null ? queueMinOffset.Value : -1L;
                    topicQueueInfo.QueueMessageCount = queue.GetMessageRealCount();
                    topicQueueInfo.QueueMaxConsumedOffset = _offsetManager.GetMinOffset(queue.Topic, queue.QueueId);
                    topicQueueInfo.Status = queue.Status;
                    topicQueueInfoList.Add(topicQueueInfo);
                }
            }

            var data = _binarySerializer.Serialize(topicQueueInfoList);
            return new RemotingResponse((int)ResponseCode.Success, request.Sequence, data);
        }

        public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
        {
            if (BrokerController.Instance.IsCleaning)
            {
                return BuildBrokerIsCleaningResponse(remotingRequest);
            }

            var request = DeserializePullMessageRequest(remotingRequest.Body);
            var topic = request.MessageQueue.Topic;
            var tags = request.Tags;
            var queueId = request.MessageQueue.QueueId;
            var pullOffset = request.QueueOffset;

            //如果消费者第一次过来拉取消息，则计算下一个应该拉取的位置，并返回给消费者
            if (pullOffset < 0)
            {
                var nextConsumeOffset = GetNextConsumeOffset(topic, queueId, request.ConsumerGroup, request.ConsumeFromWhere);
                return BuildNextOffsetResetResponse(remotingRequest, nextConsumeOffset);
            }

            //尝试拉取消息
            var pullResult = PullMessages(topic, tags, queueId, pullOffset, request.PullMessageBatchSize);

            //处理消息拉取结果
            if (pullResult.Status == PullStatus.Found)
            {
                return BuildFoundResponse(remotingRequest, pullResult.Messages);
            }
            else if (pullResult.Status == PullStatus.NextOffsetReset)
            {
                return BuildNextOffsetResetResponse(remotingRequest, pullResult.NextBeginOffset);
            }
            else if (pullResult.Status == PullStatus.QueueNotExist)
            {
                return BuildQueueNotExistResponse(remotingRequest);
            }
            else if (pullResult.Status == PullStatus.NoNewMessage)
            {
                if (request.SuspendPullRequestMilliseconds > 0)
                {
                    var pullRequest = new PullRequest(
                        remotingRequest,
                        request,
                        context,
                        DateTime.Now,
                        request.SuspendPullRequestMilliseconds,
                        ExecutePullRequest,
                        ExecutePullRequest,
                        ExecuteNoNewMessagePullRequest,
                        ExecuteReplacedPullRequest);
                    _suspendedPullRequestManager.SuspendPullRequest(pullRequest);
                    return null;
                }
                return BuildNoNewMessageResponse(remotingRequest);
            }
            else
            {
                throw new Exception("Invalid pull result status.");
            }
        }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest request)
 {
     var pullMessageRequest = _binarySerializer.Deserialize<PullMessageRequest>(request.Body);
     var messages = _messageService.GetMessages(
         pullMessageRequest.MessageQueue.Topic,
         pullMessageRequest.MessageQueue.QueueId,
         pullMessageRequest.QueueOffset,
         pullMessageRequest.PullMessageBatchSize);
     if (messages.Count() > 0)
     {
         var pullMessageResponse = new PullMessageResponse(messages);
         var responseData = _binarySerializer.Serialize(pullMessageResponse);
         return new RemotingResponse((int)PullStatus.Found, request.Sequence, responseData);
     }
     else
     {
         var pullRequest = new PullRequest(
             request.Sequence,
             pullMessageRequest,
             context,
             DateTime.Now,
             SuspendPullRequestMilliseconds,
             ExecutePullRequest,
             ExecutePullRequest,
             ExecuteReplacedPullRequest);
         _brokerController.SuspendedPullRequestManager.SuspendPullRequest(pullRequest);
         return null;
     }
 }

        public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
        {
            var request = _binarySerializer.Deserialize<QueryConsumerInfoRequest>(remotingRequest.Body);
            var consumerInfoList = new List<ConsumerInfo>();

            if (!string.IsNullOrEmpty(request.GroupName))
            {
                var consumerGroups = _consumerManager.QueryConsumerGroup(request.GroupName);
                foreach (var consumerGroup in consumerGroups)
                {
                    foreach (var topicConsumeInfo in GetConsumerInfoForGroup(consumerGroup, request.Topic))
                    {
                        consumerInfoList.Add(topicConsumeInfo);
                    }
                }
            }
            else
            {
                var consumerGroups = _consumerManager.GetAllConsumerGroups();
                foreach (var consumerGroup in consumerGroups)
                {
                    foreach (var topicConsumeInfo in GetConsumerInfoForGroup(consumerGroup, request.Topic))
                    {
                        consumerInfoList.Add(topicConsumeInfo);
                    }
                }
            }

            return RemotingResponseFactory.CreateResponse(remotingRequest, _binarySerializer.Serialize(consumerInfoList));
        }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest request)
 {
     var topic = Encoding.UTF8.GetString(request.Body);
     var queueIds = _messageService.GetQueueIdsForConsumer(topic);
     var data = Encoding.UTF8.GetBytes(string.Join(",", queueIds));
     return new RemotingResponse((int)ResponseCode.Success, request.Sequence, data);
 }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
 {
     var topic = Encoding.UTF8.GetString(remotingRequest.Body);
     var queueIds = _queueService.GetOrCreateQueues(topic, QueueStatus.Normal).Select(x => x.QueueId).ToList();
     var data = Encoding.UTF8.GetBytes(string.Join(",", queueIds));
     return new RemotingResponse((int)ResponseCode.Success, remotingRequest.Sequence, data);
 }

        public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
        {
            var request = MessageUtils.DecodeSendMessageRequest(remotingRequest.Body);
            var message = request.Message;
            var queueId = request.QueueId;
            var queue = _queueService.GetQueue(message.Topic, queueId);
            if (queue == null)
            {
                throw new QueueNotExistException(message.Topic, queueId);
            }

            //消息写文件需要加锁，确保顺序写文件
            MessageStoreResult result = null;
            lock (_syncObj)
            {
                var queueOffset = queue.NextOffset;
                var messageRecord = _messageStore.StoreMessage(queueId, queueOffset, message);
                queue.AddMessage(messageRecord.LogPosition);
                queue.IncrementNextOffset();
                result = new MessageStoreResult(messageRecord.MessageId, message.Code, message.Topic, queueId, queueOffset);
            }

            //如果需要立即通知所有消费者有新消息，则立即通知
            if (BrokerController.Instance.Setting.NotifyWhenMessageArrived)
            {
                _suspendedPullRequestManager.NotifyNewMessage(request.Message.Topic, result.QueueId, result.QueueOffset);
            }

            var data = MessageUtils.EncodeMessageStoreResult(result);
            return RemotingResponseFactory.CreateResponse(remotingRequest, data);
        }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest request)
 {
     var topic = Encoding.UTF8.GetString(request.Body);
     var queueCount = _messageService.GetTopicQueueCount(topic);
     var data = BitConverter.GetBytes(queueCount);
     return new RemotingResponse((int)ResponseCode.Success, request.Sequence, data);
 }

        public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
        {
            if (BrokerController.Instance.IsCleaning)
            {
                throw new BrokerCleanningException();
            }

            var request = _binarySerializer.Deserialize<QueryTopicQueueInfoRequest>(remotingRequest.Body);
            var topicQueueInfoList = new List<TopicQueueInfo>();
            var queues = _queueStore.QueryQueues(request.Topic).ToList().OrderBy(x => x.Topic).ThenBy(x => x.QueueId);

            foreach (var queue in queues)
            {
                var topicQueueInfo = new TopicQueueInfo();
                topicQueueInfo.Topic = queue.Topic;
                topicQueueInfo.QueueId = queue.QueueId;
                topicQueueInfo.QueueCurrentOffset = queue.NextOffset - 1;
                topicQueueInfo.QueueMinOffset = queue.GetMinQueueOffset();
                topicQueueInfo.QueueMinConsumedOffset = _offsetStore.GetMinConsumedOffset(queue.Topic, queue.QueueId);
                topicQueueInfo.ProducerVisible = queue.Setting.ProducerVisible;
                topicQueueInfo.ConsumerVisible = queue.Setting.ConsumerVisible;
                topicQueueInfoList.Add(topicQueueInfo);
            }

            return RemotingResponseFactory.CreateResponse(remotingRequest, _binarySerializer.Serialize(topicQueueInfoList));
        }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest remotingRequest)
 {
     var request = _binarySerializer.Deserialize<QueryMessageRequest>(remotingRequest.Body);
     var total = 0;
     var messages = _messageStore.QueryMessages(request.Topic, request.QueueId, request.Code, request.RoutingKey, request.PageIndex, request.PageSize, out total).ToList();
     return new RemotingResponse((int)ResponseCode.Success, remotingRequest.Sequence, _binarySerializer.Serialize(new QueryMessageResponse(total, messages)));
 }

 public RemotingResponse HandleRequest(IRequestHandlerContext context, RemotingRequest request)
 {
     var consumerData = _binarySerializer.Deserialize<ConsumerData>(request.Body);
     _brokerController.ConsumerManager.RegisterConsumer(
         consumerData.GroupName,
         new ClientChannel(consumerData.ConsumerId, context.Channel), consumerData.SubscriptionTopics);
     return null;
 }

 public ResponseFuture(RemotingRequest request, long timeoutMillis, TaskCompletionSource<RemotingResponse> requestTaskCompletionSource)
 {
     _beginTime = DateTime.Now;
     _timeoutMillis = timeoutMillis;
     _requestTaskCompletionSource = requestTaskCompletionSource;
     SendRequestSuccess = false;
     Request = request;
 }