服務(wù)端應(yīng)用層管理數(shù)據(jù)包

1、一個完整的數(shù)據(jù)包接收完畢后悔去掉head部分交由應(yīng)用層，應(yīng)用層可根據(jù)自己的需要制定應(yīng)用層協(xié)議，這里是從body部分讀取第一個字符串作為處理方法。

服務(wù)端和客戶端通用，Update需要在一個線程內(nèi)執(zhí)行

namespace SelfFramework.Network
{
    public class NetManager
    {
        private INetEventListener eventListener;
        private ThreadSynchronizationContext threadSynchronizationContext = ThreadSynchronizationContext.Instance;

        private bool isService;
        private TService service;
        private TChannel channel;

        private Stopwatch stopwatch;
        private readonly int updateTime;

        public NetManager(INetEventListener eventListener,int updateTime)
        {
            this.eventListener = eventListener;
            this.updateTime = updateTime;
            this.stopwatch = new Stopwatch();
        }

        public TService StartServer(IPEndPoint ipEndPoint)
        {
            this.service=new TService(this.threadSynchronizationContext, ipEndPoint, eventListener);
            this.isService = true;
            return this.service;
        }

        public TChannel StartConnect(IPEndPoint ipEndPoint)
        {
            if (this.channel == null || this.channel.IsUsing == false)
            {
                this.channel = new TChannel(this.threadSynchronizationContext, ipEndPoint, this.eventListener);
            }
            else
            {
                this.channel.ConnectAsync();
            }
            this.isService = false;
            return this.channel;
        }

        public void Update()
        {
            stopwatch.Start();

            threadSynchronizationContext.Update();
            if (this.isService)
            {
                this.service.Update();
            }
            else
            {
                this.channel.Update();
            }

            stopwatch.Stop();
            int sleepTime = Math.Max(0, updateTime - (int)stopwatch.ElapsedMilliseconds);
            Thread.Sleep(sleepTime);
            stopwatch.Reset();
        }
    }
}

服務(wù)端應(yīng)用層接口

namespace SelfFramework.Manager
{
    public class ServerListener:BaseBehaviour,INetEventListener
    {
        private NetManager mServer;
        private TService service;

        public void StartUp(IPEndPoint ipEndPoint, bool startHeartBeat)
        {
            this.mServer = new NetManager(this, 15);
            service = this.mServer.StartServer(ipEndPoint);
            if (startHeartBeat)
            {
                service.StartCheckHeartBeat();
            }
        }

        public void OnNetworkError(TChannel channel, SocketError socketError, SocketAsyncOperation lastOperation)
        {
            Console.WriteLine("錯誤：" + socketError + "最后的操作：>" + lastOperation);
        }

        public void OnNetworkReceive(TChannel channel, byte[] data)
        {
            networkManager.EnPacket(new NetMsg(channel, data));
        }

        public void OnPeerConnected(TChannel channel)
        {
            clientPeerManager.AddClientPeer(channel);
        }

        public void OnPeerDisconnected(TChannel channel, DisconnectInfo disconnectInfo)
        {
            clientPeerManager.RemoveClientPeer(channel.channelId);
        }

        public void StopServer()
        {
            //mServer?.Stop();
            //mServer = null;
        }

        public void Update()
        {
            mServer.Update();
        }
    }
}

應(yīng)用層網(wǎng)絡(luò)管理器

namespace SelfFramework.Manager
{
    public struct NetMsg
    {
        public TChannel channel;
        public byte[] data;
        public NetMsg(TChannel channel,byte[] data)
        {
            this.channel = channel;
            this.data = data;
        }
    }

    public class NetworkManager : BaseManager
    {
        //分發(fā)器
        private ConcurrentBag<IDispatcher> dispatcherBag;
        private ServerListener serverListener;

        private ConcurrentQueue<NetMsg> taskQueue = new ConcurrentQueue<NetMsg>();

        private static readonly object lockObj = new object();
        private readonly static log4net.ILog Log = LogHelper.GetLogger();

        public override void InitManager()
        {
            dispatcherBag = new ConcurrentBag<IDispatcher>()
            {
                new NetworkDispatcher()
            };

            this.IsUpdate = true;
            serverListener = new ServerListener();
            serverListener.StartUp(new IPEndPoint(IPAddress.Parse("127.0.0.1"), 17000), true);

            Task.Run(() =>
            {
                while (true)
                {
                    serverListener.Update();        //socket線程
                }
            });
        }

        public override void OnDispose()
        {
            taskQueue?.Clear();
        }

        public void EnPacket(NetMsg msg)
        {
            lock (lockObj)
            {
                taskQueue.Enqueue(msg);
            }
        }

        private NetMsg DePacket()
        {
            lock (lockObj)
            {
                taskQueue.TryDequeue(out NetMsg msg);
                return msg;
            }
        }

        public override void OnUpdate()
        {
            if (taskQueue.IsEmpty)
            {
                return;
            }

            try
            {
                NetMsg msg = DePacket();  //數(shù)據(jù)包出列
                foreach (IDispatcher dispatcher in dispatcherBag)
                {
                    dispatcher.OnMessage(msg.channel, msg.data);
                }
            }
            catch (Exception e)
            {
                Log.ErrorFormat("數(shù)據(jù)包處理出錯:>{0}", e.Message);
            }
        }
    }
}

服務(wù)端協(xié)議分發(fā)器，這里是讀取一個字符串，剩余部分作為數(shù)據(jù)，可自行制定應(yīng)用層協(xié)議

namespace SelfFramework.Manager
{
    public sealed class NetworkDispatcher : BaseBehaviour,IDispatcher
    {
        private ConcurrentDictionary<string, HandlerDelegate> handleTable;

        public NetworkDispatcher()
        {
            handleTable = new ConcurrentDictionary<string, HandlerDelegate>();
            RegistHandlers();
        }

        public void OnMessage(TChannel channel, byte[] body)
        {
            NetDataReader reader = new NetDataReader(body, 0, body.Length);
            string protocol = reader.GetString();
            byte[] data = null;
            if (reader.ReadPos < reader.Capacity)
            {
                data = reader.GetBytes();
            }

            if (handleTable.TryGetValue(protocol, out HandlerDelegate handler))
            {
                ClientPeer clientPeer = clientPeerManager.GetClientPeer(channel.channelId);
                if (clientPeer!=null)
                {
                    handler.Invoke(clientPeer, data);
                }
            }
        }

        private void RegistHandlers()
        {
            Type parentType = typeof(IHandler);
            Type[] assemblyTypes = parentType.Assembly.GetTypes();
            for (int i = 0; i < assemblyTypes.Length; i++)
            {
                Type type = assemblyTypes[i];
                if (type.IsAbstract || type.IsInterface)
                {
                    continue;
                }
                if (parentType.IsAssignableFrom(type) == false)
                {
                    continue;
                }

                IHandler handler = Activator.CreateInstance(type) as IHandler;
                MethodInfo[] methodInfos = handler.GetType().GetMethods();
                for (int j = 0; j < methodInfos.Length; j++)
                {
                    MethodInfo methodInfo = methodInfos[j];
                    HandlerAttribute attribute = methodInfo.GetCustomAttribute(typeof(HandlerAttribute)) as HandlerAttribute;
                    if (attribute != null)
                    {
                        HandlerDelegate handlerDelegate = methodInfo.CreateDelegate(typeof(HandlerDelegate), handler) as HandlerDelegate;
                        RegistHandler(attribute.protocol, handlerDelegate);
                    }
                }
            }
        }

        private void RegistHandler(string cmd, HandlerDelegate handler)
        {
            if (handleTable.ContainsKey(cmd))
            {
                handleTable[cmd] = handler;
            }
            else
            {
                handleTable.TryAdd(cmd, handler);
            }
        }

        private void UnregistHandler(string protocol)
        {
            handleTable.TryRemove(protocol, out HandlerDelegate handler);
        }
    }
}

一個測試?yán)?/em>

namespace SelfFramework.Handler
{
    public sealed class TestHandler : IHandler
    {
        [Handler(Protocol.TestFrame)]
        public void ResponseTestFrame(ClientPeer clientPeer, byte[] data)
        {

        }

        [Handler(Protocol.Test)]
        public void ResponseTest(ClientPeer clientPeer, byte[] data)
        {
            NetDataReader dataReader = new NetDataReader(data, 0, data.Length);
            Console.WriteLine(dataReader.GetInt32());
            clientPeer.Send(Protocol.Test, new byte[] { 123, 0, 0, 0 });
        }

    }
}