DotNetty系列五：使用委托和事件调整代码模式，将群发改为ConcurrentDictionary

在上一节基础上，使用委托和事件调整代码模式。

1.将解码器，编码器，去除多余代码。

    public class CommonEncoder<T> : MessageToByteEncoder<T>{protected override void Encode(IChannelHandlerContext context, T message, IByteBuffer output){//序列化类byte[] messageBytes = MessagePackHelper.SerializeToBinary(message);IByteBuffer initialMessage = Unpooled.Buffer(messageBytes.Length);initialMessage.WriteBytes(messageBytes);output.WriteBytes(initialMessage);}}public class CommonDecoder : ByteToMessageDecoder{protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output){byte[] array = new byte[input.ReadableBytes];input.GetBytes(input.ReaderIndex, array, 0, input.ReadableBytes);input.Clear();var temp = MessagePackHelper.DeserializeWithBinary<Message>(array);output.Add(temp);}}

2.在类库项目中新建委托类。

    public delegate void CommandReceiveEvent(object sender, MessageEventArgs e);public delegate void MessageSendEvent(object sender, MessageEventArgs e);public delegate void MessageGroupSendEvent(object sender, MessageEventArgs e);public class MessageEventArgs : EventArgs{public MessageEventArgs(IMessage cmd) => CMD = cmd;public IMessage CMD { set; get; }}

3.客户端主程序里修改：

                        pipeline.AddLast(new CommonEncoder<Message>());pipeline.AddLast(new CommonDecoder());pipeline.AddLast(new IdleStateHandler(0, 10, 0));HelloClientHandler ClientHandler = new HelloClientHandler{_Socket = pipeline.FirstContext()};ClientHandler.MessageReceived += ClientHandler_MessageReceived;ClientHandler.MessageSend += ClientHandler_MessageSend;pipeline.AddLast(ClientHandler);private async static void ClientHandler_MessageSend(object sender, MessageEventArgs e){try{var client = sender as HelloClientHandler;await Task.Run(() => { client.SendData(e.CMD); });}catch (Exception ex) { }}private async static void ClientHandler_MessageReceived(object sender, MessageEventArgs e){try{var client = sender as HelloClientHandler;await Task.Run(() =>{switch (e.CMD.Command){case COMMAND.Message:Console.WriteLine($"Decoder方式，从服务端接收:{e.CMD.Content}");return;default:return;}});}catch (Exception ex) { }}

HelloClientHandler里修改:

        public event CommandReceiveEvent MessageReceived;public event MessageSendEvent MessageSend;public async void SendData<T>(T obj){try{await _Socket.WriteAndFlushAsync(obj);}catch (Exception ex) { }}private void OnMessageReceive(Message msg) => MessageReceived?.Invoke(this, new MessageEventArgs(msg));private void OnMessageSend(Message msg) => MessageSend?.Invoke(this, new MessageEventArgs(msg));public override async void UserEventTriggered(IChannelHandlerContext context, object evt){await Task.Run(() =>{Console.WriteLine("客户端循环心跳监测发送: " + DateTime.Now);if (evt is IdleStateEvent eventState){if (eventState.State == IdleState.WriterIdle){Message ms = new Message { Command = COMMAND.HeartBeat, Content = $"biubiu:{DateTime.Now}" };OnMessageSend(ms);}}});}public override async void ChannelActive(IChannelHandlerContext context){_Socket = context;Console.WriteLine("我是客户端.");Console.WriteLine($"连接至服务端{context}.");await Task.Run(() =>{Message ms = new Message { Command = COMMAND.Message, Content = $"客户端{(context.Channel.LocalAddress as IPEndPoint).Port}" };OnMessageSend(ms);});}public override async void ChannelRead(IChannelHandlerContext context, object message){await Task.Run(() =>{if (message is Message oo){OnMessageReceive(oo);}});}

4.服务端主程序里修改：

                        pipeline.AddLast(new CommonEncoder<Message>());pipeline.AddLast(new CommonDecoder());pipeline.AddLast(new IdleStateHandler(150, 0, 0));HelloServerHandler ServerHandler = new HelloServerHandler{_Socket = pipeline.FirstContext()};ServerHandler.MessageReceived += ServerHandler_MessageReceived;ServerHandler.MessageSend += ServerHandler_MessageSend;ServerHandler.MessageGroupSend += ServerHandler_MessageGroupSend;pipeline.AddLast(ServerHandler);private async static void ServerHandler_MessageGroupSend(object sender, MessageEventArgs e){try{var clients = sender as HelloServerHandler;await Task.Run(() => { clients.GroupSendData(e.CMD); });}catch (Exception ex) { }}private async static void ServerHandler_MessageSend(object sender, MessageEventArgs e){try{var client = sender as HelloServerHandler;await Task.Run(() => { client.SendData(e.CMD); });}catch (Exception ex) { }}private async static void ServerHandler_MessageReceived(object sender, MessageEventArgs e){try{var client = sender as HelloServerHandler;await Task.Run(() =>{switch (e.CMD.Command){case COMMAND.Message:Console.WriteLine($"解码器方式，从客户端接收:{e.CMD.Content}:{DateTime.Now}");ServerHandler_MessageSend(sender, new MessageEventArgs(new Message { Command = COMMAND.Message, Content = "服务端接收到后返回，我是服务端" }));return;default:Console.WriteLine($"解码器方式，客户端心跳:{e.CMD.Content}");ServerHandler_MessageSend(sender, new MessageEventArgs(new Message { Command = COMMAND.Message, Content = "服务端接收到心跳连接" }));return;}});}catch (Exception ex) { }}

HelloServerHandler里修改:

        public event CommandReceiveEvent MessageReceived;public event MessageSendEvent MessageSend;public event MessageGroupSendEvent MessageGroupSend;public async void SendData<T>(T obj){try{await _Socket.WriteAndFlushAsync(obj);}catch (Exception ex) { }}public void GroupSendData<T>(T obj){try{AllClients.Values.ToList().ForEach(async s => await s.WriteAndFlushAsync(obj));}catch (Exception ex) { }}private void OnMessageReceive(Message msg) => MessageReceived?.Invoke(this, new MessageEventArgs(msg));private void OnMessageSend(Message msg) => MessageSend?.Invoke(this, new MessageEventArgs(msg));private void OnMessageGroupSend(Message msg) => MessageGroupSend?.Invoke(this, new MessageEventArgs(msg));public override async void ChannelRead(IChannelHandlerContext context, object message){await Task.Run(() =>{if (message is Message oo){OnMessageReceive(oo);}});}static volatile ConcurrentDictionary<int, IChannel> AllClients = new ConcurrentDictionary<int, IChannel>();//客户端连接进来时public override async void HandlerAdded(IChannelHandlerContext context){await Task.Run(() =>{Console.WriteLine($"客户端{context}上线.");base.HandlerAdded(context);AllClients.AddOrUpdate((context.Channel.RemoteAddress as IPEndPoint).Port, context.Channel, (k, v) => v);Message ms = new Message { Command = COMMAND.Message, Content = $"欢迎{context.Channel.RemoteAddress}加入." };OnMessageGroupSend(ms);});}//客户端下线断线时public override async void HandlerRemoved(IChannelHandlerContext context){await Task.Run(() =>{Console.WriteLine($"客户端{context}下线.");base.HandlerRemoved(context);AllClients.TryRemove((context.Channel.RemoteAddress as IPEndPoint).Port, out IChannel temp);Message ms = new Message { Command = COMMAND.Message, Content = $"恭送{context.Channel.RemoteAddress}离开." };OnMessageGroupSend(ms);});}//服务器监听到客户端活动时public override async void ChannelActive(IChannelHandlerContext context){_Socket = context;//赋值await Task.Run(() =>{Console.WriteLine($"客户端{context.Channel.RemoteAddress}在线.");base.ChannelActive(context);});}

这样就使用，委托和事件模式，将群发改为ConcurrentDictionary，实现了上一节一样的效果。

项目下载地址：项目下载