完善心跳的设计。

Author: tangxuehua

src/EQueue.AdminWeb/Controllers/HomeController.cs

@@ -22,6 +22,7 @@ public ActionResult Index()
                 TopicCount = result.TopicCount,
                 QueueCount = result.QueueCount,
                 ConsumerGroupCount = result.ConsumerGroupCount,
+                ProducerCount = result.ProducerCount,
                 ConsumerCount = result.ConsumerCount,
                 TotalUnConsumedMessageCount = result.TotalUnConsumedMessageCount,
                 MessageChunkCount = result.MessageChunkCount,
@@ -48,6 +49,14 @@ public ActionResult QueueInfo(string topic)
                 TopicQueueInfos = topicQueueInfos
             });
         }
+        public ActionResult ProducerInfo(string group, string topic)
+        {
+            var producerIds = _messageService.GetProducerInfo();
+            return View(new ProducerViewModel
+            {
+                ProducerIds = producerIds
+            });
+        }
         public ActionResult ConsumerInfo(string group, string topic)
         {
             var consumerInfos = _messageService.GetConsumerInfo(group, topic);

src/EQueue.AdminWeb/EQueue.AdminWeb.csproj

@@ -150,6 +150,7 @@
       <DependentUpon>Global.asax</DependentUpon>
     </Compile>
     <Compile Include="Models\BrokerStatisticInfoViewModel.cs" />
+    <Compile Include="Models\ProducerViewModel.cs" />
     <Compile Include="Models\MessageViewModel.cs" />
     <Compile Include="Models\ConsumerViewModel.cs" />
     <Compile Include="Models\TopicQueueViewModel.cs" />
@@ -190,6 +191,7 @@
     <Content Include="Views\Shared\_Layout.cshtml" />
     <Content Include="Views\Home\Index.cshtml" />
     <Content Include="Views\Home\ConsumerInfo.cshtml" />
+    <Content Include="Views\Home\ProducerInfo.cshtml" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\EQueue\EQueue.csproj">

src/EQueue.AdminWeb/MessageService.cs

@@ -13,6 +13,7 @@ namespace EQueue.AdminWeb
 {
     public class MessageService
     {
+        private readonly byte[] EmptyBytes = new byte[0];
         private readonly SocketRemotingClient _remotingClient;
         private readonly IBinarySerializer _binarySerializer;
         private readonly IScheduleService _scheduleService;
@@ -82,7 +83,21 @@ public IEnumerable<TopicQueueInfo> GetTopicQueueInfo(string topic)
             }
             else
             {
-                throw new Exception(string.Format("QueryTopicQueueInfo failed, errorMessage: {0}", Encoding.UTF8.GetString(remotingResponse.Body)));
+                throw new Exception(string.Format("GetTopicQueueInfo failed, errorMessage: {0}", Encoding.UTF8.GetString(remotingResponse.Body)));
+            }
+        }
+        public IEnumerable<string> GetProducerInfo()
+        {
+            var remotingRequest = new RemotingRequest((int)RequestCode.QueryProducerInfo, EmptyBytes);
+            var remotingResponse = _remotingClient.InvokeSync(remotingRequest, 10000);
+            if (remotingResponse.Code == (int)ResponseCode.Success)
+            {
+                var producerIds = Encoding.UTF8.GetString(remotingResponse.Body);
+                return producerIds.Split(new[] { "," }, StringSplitOptions.RemoveEmptyEntries);
+            }
+            else
+            {
+                throw new Exception(string.Format("GetProducerInfo failed, errorMessage: {0}", Encoding.UTF8.GetString(remotingResponse.Body)));
             }
         }
         public IEnumerable<ConsumerInfo> GetConsumerInfo(string group, string topic)
@@ -96,7 +111,7 @@ public IEnumerable<ConsumerInfo> GetConsumerInfo(string group, string topic)
             }
             else
             {
-                throw new Exception(string.Format("QueryConsumerInfo failed, errorMessage: {0}", Encoding.UTF8.GetString(remotingResponse.Body)));
+                throw new Exception(string.Format("GetConsumerInfo failed, errorMessage: {0}", Encoding.UTF8.GetString(remotingResponse.Body)));
             }
         }
         public void AddQueue(string topic)

src/EQueue.AdminWeb/Models/BrokerStatisticInfoViewModel.cs

@@ -8,6 +8,9 @@ public class BrokerStatisticInfoViewModel
         /// <summary>队列个数
         /// </summary>
         public int QueueCount { get; set; }
+        /// <summary>生产者个数
+        /// </summary>
+        public int ProducerCount { get; set; }
         /// <summary>消费者组个数
         /// </summary>
         public int ConsumerGroupCount { get; set; }

src/EQueue.AdminWeb/Models/ProducerViewModel.cs

@@ -0,0 +1,10 @@
+using System.Collections.Generic;
+using EQueue.Protocols;
+
+namespace EQueue.AdminWeb.Models
+{
+    public class ProducerViewModel
+    {
+        public IEnumerable<string> ProducerIds { get; set; }
+    }
+}

src/EQueue.AdminWeb/Views/Home/ConsumerInfo.cshtml

@@ -65,6 +65,7 @@
                     </td>
                     <td class="text-center vertical-center">
                         @(item.QueueCurrentOffset - item.ConsumedOffset)
+                    </td>
                 </tr>
             }
         </tbody>

src/EQueue.AdminWeb/Views/Home/Index.cshtml

@@ -24,6 +24,12 @@
                 @Model.QueueCount
             </div>
         </div>
+        <div class="form-group">
+            <label class="col-sm-2 control-label">生产者个数：</label>
+            <div class="col-sm-10">
+                @Model.ProducerCount
+            </div>
+        </div>
         <div class="form-group">
             <label class="col-sm-2 control-label">消费者组个数：</label>
             <div class="col-sm-10">

src/EQueue.AdminWeb/Views/Home/ProducerInfo.cshtml

@@ -0,0 +1,25 @@
+@model EQueue.AdminWeb.Models.ProducerViewModel
+@{
+    ViewBag.Title = "生产者列表";
+}
+<div>
+    <table class="table table-striped">
+        <thead>
+            <tr>
+                <th class="vertical-center">
+                    ProducerId
+                </th>
+            </tr>
+        </thead>
+        <tbody>
+            @foreach (var item in Model.ProducerIds)
+            {
+                <tr>
+                    <td class="vertical-center">
+                        @item
+                    </td>
+                </tr>
+            }
+        </tbody>
+    </table>
+</div>

src/EQueue.AdminWeb/Views/Shared/_Layout.cshtml

@@ -22,6 +22,7 @@
                 </button>
                 @Html.ActionLink("首页", "Index", "Home", null, new { @class = "navbar-brand" })
                 @Html.ActionLink("队列信息", "QueueInfo", "Home", null, new { @class = "navbar-brand" })
+                @Html.ActionLink("生产者列表", "ProducerInfo", "Home", null, new { @class = "navbar-brand" })
                 @Html.ActionLink("消费者列表", "ConsumerInfo", "Home", null, new { @class = "navbar-brand" })
                 @Html.ActionLink("消息查询", "Message", "Home", null, new { @class = "navbar-brand" })
             </div>

src/EQueue/Broker/BrokerController.cs

@@ -26,6 +26,7 @@ public class BrokerController
         private readonly IMessageStore _messageStore;
         private readonly IConsumeOffsetStore _consumeOffsetStore;
         private readonly SuspendedPullRequestManager _suspendedPullRequestManager;
+        private readonly ProducerManager _producerManager;
         private readonly ConsumerManager _consumerManager;
         private readonly SocketRemotingServer _producerSocketRemotingServer;
         private readonly SocketRemotingServer _consumerSocketRemotingServer;
@@ -36,6 +37,10 @@ public class BrokerController
         private int _isCleaning = 0;
 
         public BrokerSetting Setting { get; private set; }
+        public ProducerManager ProducerManager
+        {
+            get { return _producerManager; }
+        }
         public ConsumerManager ConsumerManager
         {
             get { return _consumerManager; }
@@ -52,6 +57,7 @@ public static BrokerController Instance
         private BrokerController(BrokerSetting setting)
         {
             Setting = setting ?? new BrokerSetting();
+            _producerManager = ObjectContainer.Resolve<ProducerManager>();
             _consumerManager = ObjectContainer.Resolve<ConsumerManager>();
             _messageStore = ObjectContainer.Resolve<IMessageStore>();
             _consumeOffsetStore = ObjectContainer.Resolve<IConsumeOffsetStore>();
@@ -64,6 +70,7 @@ private BrokerController(BrokerSetting setting)
             _adminSocketRemotingServer = new SocketRemotingServer("EQueue.Broker.AdminRemotingServer", Setting.AdminAddress, Setting.SocketSetting);
 
             _logger = ObjectContainer.Resolve<ILoggerFactory>().Create(GetType().FullName);
+            _producerSocketRemotingServer.RegisterConnectionEventListener(new ProducerConnectionEventListener(this));
             _consumerSocketRemotingServer.RegisterConnectionEventListener(new ConsumerConnectionEventListener(this));
             RegisterRequestHandlers();
 
@@ -150,6 +157,7 @@ public BrokerController Start()
             _consumeOffsetStore.Start();
             _messageStore.Start();
             _queueStore.Start();
+            _producerManager.Start();
             _consumerManager.Start();
             _suspendedPullRequestManager.Start();
             _consumerSocketRemotingServer.Start();
@@ -172,6 +180,7 @@ public BrokerController Shutdown()
                 _producerSocketRemotingServer.Shutdown();
                 _consumerSocketRemotingServer.Shutdown();
                 _adminSocketRemotingServer.Shutdown();
+                _producerManager.Shutdown();
                 _consumerManager.Shutdown();
                 _suspendedPullRequestManager.Shutdown();
                 _messageStore.Shutdown();
@@ -189,6 +198,7 @@ public BrokerStatisticInfo GetBrokerStatisticInfo()
             statisticInfo.QueueCount = _queueStore.GetAllQueueCount();
             statisticInfo.TotalUnConsumedMessageCount = _queueStore.GetTotalUnConusmedMessageCount();
             statisticInfo.ConsumerGroupCount = _consumerManager.GetConsumerGroupCount();
+            statisticInfo.ProducerCount = _producerManager.GetProducerCount();
             statisticInfo.ConsumerCount = _consumerManager.GetConsumerCount();
             statisticInfo.MessageChunkCount = _messageStore.ChunkCount;
             statisticInfo.MessageMinChunkNum = _messageStore.MinChunkNum;
@@ -209,20 +219,22 @@ private void RemoveNotExistQueueConsumeOffsets()
         }
         private void RegisterRequestHandlers()
         {
+            _producerSocketRemotingServer.RegisterRequestHandler((int)RequestCode.ProducerHeartbeat, new ProducerHeartbeatRequestHandler(this));
             _producerSocketRemotingServer.RegisterRequestHandler((int)RequestCode.SendMessage, new SendMessageRequestHandler(this));
 
+            _consumerSocketRemotingServer.RegisterRequestHandler((int)RequestCode.ConsumerHeartbeat, new ConsumerHeartbeatRequestHandler(this));
             _consumerSocketRemotingServer.RegisterRequestHandler((int)RequestCode.PullMessage, new PullMessageRequestHandler());
 
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.GetTopicQueueIdsForProducer, new GetTopicQueueIdsForProducerRequestHandler());
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.GetTopicQueueIdsForConsumer, new GetTopicQueueIdsForConsumerRequestHandler());
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.QueryGroupConsumer, new QueryConsumerRequestHandler());
-            _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.ConsumerHeartbeat, new ConsumerHeartbeatRequestHandler(this));
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.UpdateQueueOffsetRequest, new UpdateQueueOffsetRequestHandler());
 
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.QueryBrokerStatisticInfo, new QueryBrokerStatisticInfoRequestHandler());
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.CreateTopic, new CreateTopicRequestHandler());
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.DeleteTopic, new DeleteTopicRequestHandler());
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.QueryTopicQueueInfo, new QueryTopicQueueInfoRequestHandler());
+            _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.QueryProducerInfo, new QueryProducerInfoRequestHandler());
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.QueryConsumerInfo, new QueryConsumerInfoRequestHandler());
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.AddQueue, new AddQueueRequestHandler());
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.DeleteQueue, new DeleteQueueRequestHandler());
@@ -231,6 +243,24 @@ private void RegisterRequestHandlers()
             _adminSocketRemotingServer.RegisterRequestHandler((int)RequestCode.GetMessageDetail, new GetMessageDetailRequestHandler());
         }
 
+        class ProducerConnectionEventListener : IConnectionEventListener
+        {
+            private BrokerController _brokerController;
+
+            public ProducerConnectionEventListener(BrokerController brokerController)
+            {
+                _brokerController = brokerController;
+            }
+
+            public void OnConnectionAccepted(ITcpConnection connection) { }
+            public void OnConnectionEstablished(ITcpConnection connection) { }
+            public void OnConnectionFailed(SocketError socketError) { }
+            public void OnConnectionClosed(ITcpConnection connection, SocketError socketError)
+            {
+                var producerId = ClientIdFactory.CreateClientId(connection.RemotingEndPoint as IPEndPoint);
+                _brokerController._producerManager.RemoveProducer(producerId);
+            }
+        }
         class ConsumerConnectionEventListener : IConnectionEventListener
         {
             private BrokerController _brokerController;