hjha-server/GameNetModule/EventDispatcher/MessageDispatcher.cs

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);

    /// <summary>
    /// 网络消息 分发器
    /// </summary>
    public sealed class MessageDispatcher : SingleInstance<MessageDispatcher>
    {
        /// <summary>
        /// 是否需要按顺序执行  一般来说子游戏需要按顺序执行，中心服务器不需要按顺序等待
        /// </summary>
        public bool IsOrderMessage { get; private set; }

        private ConcurrentQueue<GamePacket> packes = new ConcurrentQueue<GamePacket>();

        public void AddPack(GamePacket gamePacket)
        {
            //Debug.Info($"添加一个待处理包裹:{gamePacket.OpCode}");
            packes.Enqueue(gamePacket);
        }

        private CoroutineLockManager coroutineLockManager;

        private Dictionary<int, MessageDispatcherDelegate> listeners =
            new Dictionary<int, MessageDispatcherDelegate>();

        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);
                }
            }
        }

        /// <summary>
        /// 按玩家的顺序来处理
        /// </summary>
        /// <returns></returns>
        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();
            }
        }
    }
}