using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Threading.Tasks; using MrWu.Debug; using Server.Core; namespace Server.Net { public delegate Task MessageDispatcherDelegate(GamePacket gamePacket); /// /// 网络消息 分发器 /// public sealed class MessageDispatcher : SingleInstance { /// /// 是否需要按顺序执行 一般来说子游戏需要按顺序执行,中心服务器不需要按顺序等待 /// public bool IsOrderMessage { get; private set; } private ConcurrentQueue packes = new ConcurrentQueue(); public void AddPack(GamePacket gamePacket) { //Debug.Info($"添加一个待处理包裹:{gamePacket.OpCode}"); packes.Enqueue(gamePacket); } private CoroutineLockManager coroutineLockManager; private Dictionary listeners = new Dictionary(); public void Start(bool isOrderMessage, CoroutineLockManager coroutineLockManager) { this.IsOrderMessage = isOrderMessage; this.coroutineLockManager = coroutineLockManager; } public void AddListener(int msgId, MessageDispatcherDelegate messageDispatcherDelegate) { listeners[msgId] = messageDispatcherDelegate; } public void RemoveListener(int msgId, MessageDispatcherDelegate messageDispatcherDelegate) { if (listeners.ContainsKey(msgId) && listeners[msgId] == messageDispatcherDelegate) { listeners.Remove(msgId); } } private const int MaxCnt = 200; private int PackCnt; //此处可以优化 public void Handle() { PackCnt = packes.Count; if (PackCnt > MaxCnt) { Debug.ImportantLog($"单帧处理消息过多! {packes.Count}"); PackCnt = MaxCnt; //进行统计 } while (PackCnt -- > 0) { if (!packes.TryDequeue(out GamePacket pack)) { break; } if (!listeners.TryGetValue(pack.OpCode, out var handle)) { Debug.Error($"这个客户端消息没有处理器! {pack.OpCode}"); pack.Dispose(); continue; } if (IsOrderMessage) { _ = SafeInvokeOrderMessage(pack, handle); // //同用户的消息顺序才需要等待上一个包的完成 // using (await coroutineLockManager.Wait(CoroutineLockType.ClientMessage, pack.Session.Id)) // { // await SafeInvoke(pack, handle); // } } else { _ = SafeInvoke(pack, handle); } } } /// /// 按玩家的顺序来处理 /// /// private async Task SafeInvokeOrderMessage(GamePacket packet, MessageDispatcherDelegate handler) { using (await coroutineLockManager.Wait(CoroutineLockType.ClientMessage, packet.Session.InstanceId)) { await SafeInvoke(packet, handler); } } private async Task SafeInvoke(GamePacket packet, MessageDispatcherDelegate handler) { try { if (packet.Session.IsDisConnected) { return; } long key = Debug.StartTiming(); await handler(packet); double _runtime = Debug.GetRunTime(key); if (_runtime > 1000f) { Debug.Warning($"消息执行处理慢:{packet.OpCode} {_runtime}"); } } catch (Exception e) { Debug.Error($"消息处理异常:{packet.OpCode} {e.Message}"); } finally { await PacketDispose(packet); } } private async Task PacketDispose(GamePacket packet) { try { await packet.Wait(); } finally { packet.Dispose(); } } } }