完善一下Netty项目的队列demo

simples/JT808.DotNetty.SimpleClient/Program.cs

@@ -33,7 +33,7 @@ namespace JT808.DotNetty.SimpleClient
                 services.AddJT808Configure()
                         .AddJT808Client();
                 services.AddHostedService<UpService>();
-                //services.AddHostedService<Up2019Service>();
+                services.AddHostedService<Up2019Service>();
             });
             await serverHostBuilder.RunConsoleAsync();
         }

simples/JT808.DotNetty.SimpleQueueService/Impl/JT808DotNettyReplyMessageServiceInherited.cs

@@ -0,0 +1,41 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using JT808.DotNetty.Abstractions;
+using JT808.DotNetty.Kafka;
+using JT808.DotNetty.ReplyMessage;
+using JT808.Protocol;
+using JT808.Protocol.Extensions;
+using Microsoft.Extensions.Logging;
+
+namespace JT808.DotNetty.SimpleQueueService.Impl
+{
+    public class JT808DotNettyReplyMessageServiceInherited : JT808DotNettyReplyMessageService
+    {
+        public readonly ILogger<JT808DotNettyReplyMessageServiceInherited> logger;
+
+        public JT808DotNettyReplyMessageServiceInherited(IJT808Config jT808Config,
+            IJT808MsgReplyProducer jT808MsgReplyProducer,
+            ILoggerFactory loggerFactory) 
+            : base(jT808Config, jT808MsgReplyProducer)
+        {
+            logger = loggerFactory.CreateLogger<JT808DotNettyReplyMessageServiceInherited>();
+        }
+
+        public override void Processor((string TerminalNo, byte[] Data) parameter)
+        {
+            logger.LogDebug($"{parameter.TerminalNo}:{parameter.Data.ToHexString()}");
+            base.Processor(parameter);
+        }
+
+        public override byte[] Msg0x0200(JT808HeaderPackage request)
+        {
+            logger.LogWarning("===========================================");
+            logger.LogWarning($"{request.Header.TerminalPhoneNo}---{request.OriginalData.ToArray().ToHexString()}");
+            logger.LogWarning("===========================================");
+            return base.Msg0x0200(request);
+        }
+    }
+}

simples/JT808.DotNetty.SimpleQueueService/Program.cs

@@ -1,5 +1,6 @@
 using JT808.DotNetty.Kafka;
 using JT808.DotNetty.ReplyMessage;
+using JT808.DotNetty.SimpleQueueService.Impl;
 using JT808.Protocol;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
@@ -28,8 +29,9 @@ namespace JT808.DotNetty.SimpleQueueService
                     services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
                     services.AddJT808Configure()
                             .AddJT808ClientKafka()
-                            .AddMsgReplyConsumer(hostContext.Configuration)
-                            .AddMsgReplyConsumer(hostContext.Configuration)
+                            .AddMsgReplyProducer(hostContext.Configuration)
+                            .AddMsgConsumer(hostContext.Configuration)
+                            .AddInprocJT808ReplyMessage<JT808DotNettyReplyMessageServiceInherited>();
                             ;
                 });