hjha-server/ServerCore/NetWork/TService.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using MrWu.Debug;
using NetWorkMessage;

namespace Server.Net
{
    public enum TcpOp
    {
        StartSend,
        StartRecv,
        Connect,
    }

    public struct TArgs
    {
        public TcpOp Op;
        public long ChannelId;
        public SocketAsyncEventArgs SocketAsyncEventArgs;
    }

    public enum ServiceType
    {
        Router,
        Actor,
    }

    public sealed class TService : AService
    {
        private readonly Dictionary<long, TChannel> idChannels = new Dictionary<long, TChannel>();

        private readonly SocketAsyncEventArgs innArgs = new SocketAsyncEventArgs();

        /// <summary>
        /// 监听者
        /// </summary>
        private Socket acceptor;

        public ConcurrentQueue<TArgs> Queue = new ConcurrentQueue<TArgs>();


        public TService(ServiceType serviceType)
        {
            this.ServiceType = serviceType;
        }

        public TService(IPEndPoint ipEndPoint, ServiceType serviceType)
        {
            this.ServiceType = serviceType;
            this.acceptor = new Socket(ipEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            // 容易出问题，先注释掉，按需开启
            //this.acceptor.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
            this.innArgs.Completed += this.OnComplete;
            try
            {
                this.acceptor.Bind(ipEndPoint);
            }
            catch (Exception e)
            {
                throw new Exception($"bind error: {ipEndPoint}", e);
            }

            //设置连接请求队列的最大长度，不是只能有1000个链接
            this.acceptor.Listen(1000);

            this.AcceptAsync();
        }

        private void OnComplete(object sender, SocketAsyncEventArgs e)
        {
            switch (e.LastOperation)
            {
                case SocketAsyncOperation.Accept:
                    this.Queue.Enqueue(new TArgs() { SocketAsyncEventArgs = e });
                    break;
                default:
                    throw new Exception($"socket error:{e.LastOperation}");
            }
        }

        private void OnAcceptComplete(SocketError socketError, Socket acceptSocket)
        {
            if (this.acceptor == null)
            {
                return;
            }

            if (socketError != SocketError.Success)
            {
                Debug.Error($"accept error: {socketError}");
                this.AcceptAsync();
                return;
            }

            //有客户端连接成功
            try
            {
                //创建Channel
                long id = NetServices.CreateAcceptChannelId();
                TChannel channel = new TChannel(id, acceptSocket, this);
                this.idChannels.Add(channel.Id, channel);
                long channelId = channel.Id;

                this.AcceptCallBack(channelId, channel.RemoteAddress);
            }
            catch (Exception e)
            {
                Debug.Error($"TServiceCreateChannel:{e}");
            }

            this.AcceptAsync();
        }

        private void AcceptAsync()
        {
            this.innArgs.AcceptSocket = null;
            if (this.acceptor.AcceptAsync(this.innArgs))
            {
                return;
            }

            OnAcceptComplete(this.innArgs.SocketError, this.innArgs.AcceptSocket);
        }

        public override void Create(long id, IPEndPoint ipEndPoint)
        {
            if (this.idChannels.TryGetValue(id, out TChannel _))
            {
                Debug.Error($"添加重复Channel:{id}");
                return;
            }
            
            TChannel channel = new TChannel(id, ipEndPoint, this);
            this.idChannels.Add(channel.Id, channel);
        }

        public TChannel Get(long id)
        {
            TChannel channel = null;
            this.idChannels.TryGetValue(id, out channel);
            return channel;
        }

        public override void Dispose()
        {
            Debug.Info("TChannel Dispose! --- ");
            
            this.acceptor?.Close();
            this.acceptor = null;
            this.innArgs.Dispose();

            foreach (var id in this.idChannels.Keys.ToArray())
            {
                TChannel channel = this.idChannels[id];
                channel.Dispose();
            }

            this.idChannels.Clear();
        }

        public override void Remove(long id, int error = 0)
        {
            if (this.idChannels.TryGetValue(id, out TChannel channel))
            {
                channel.Error = error;
                channel.Dispose();
            }

            this.idChannels.Remove(id);
        }

        public override void Send(long channelId, MemoryBuffer memoryBuffer)
        {
            #if DEBUG
            ServiceThreadManager.Instance.CheckThread(this,Thread.CurrentThread.ManagedThreadId);
            #endif            
            try
            {
                TChannel channel = this.Get(channelId);
                if (channel == null)
                {
                    //Debug.Log("channel is null!");
                    this.ErrorCallBack(channelId, NetErrorCode.ERR_SendMessageNotFoundTChannel);
                    return;
                }

                //Debug.Log("TChannel Send!");
                channel.Send(memoryBuffer);
            }
            catch (Exception e)
            {
                Debug.Error($"TService Send:{e}");
            }
        }
        
        public override void Update()
        {
            while (true)
            {
                if (!this.Queue.TryDequeue(out var result))
                {
                    break;
                }

                SocketAsyncEventArgs e = result.SocketAsyncEventArgs;

                if (e == null)
                {
                    switch (result.Op)
                    {
                        case TcpOp.StartSend:
                        {
                            //Debug.Log("开始发送!");
                            TChannel channel = this.Get(result.ChannelId);
                            if (channel != null)
                            {
                                channel.StartSend();
                            }

                            break;
                        }
                        case TcpOp.StartRecv:
                        {
                            TChannel channel = this.Get(result.ChannelId);
                            if (channel != null)
                            {
                                channel.StartRecv();
                            }
                            break;
                        }
                        case TcpOp.Connect:
                        {
                            Debug.Log("链接!");
                            TChannel channel = this.Get(result.ChannelId);
                            if (channel != null)
                            {
                                channel.ConnectAsync();
                            }
                            break;
                        }
                    }

                    continue;
                }

                switch (e.LastOperation)
                {
                    //监听
                    case SocketAsyncOperation.Accept:
                    {
                        SocketError socketError = e.SocketError;
                        Socket acceptSocket = e.AcceptSocket;
                        this.OnAcceptComplete(socketError, acceptSocket);

                        break;
                    }
                    //链接
                    case SocketAsyncOperation.Connect:
                    {
                        TChannel channel = this.Get(result.ChannelId);
                        if (channel != null)
                        {
                            channel.OnConnectComplete(e);
                        }

                        break;
                    }
                    //断开
                    case SocketAsyncOperation.Disconnect:
                    {
                        TChannel channel = this.Get(result.ChannelId);
                        if (channel != null)
                        {
                            channel.OnDisconnectComplete(e);
                        }

                        break;
                    }

                    //接收
                    case SocketAsyncOperation.Receive:
                    {
                        TChannel channel = this.Get(result.ChannelId);
                        if (channel != null)
                        {
                            channel.OnRecvComplete(e);
                        }

                        break;
                    }
                    //发送
                    case SocketAsyncOperation.Send:
                    {
                        TChannel channel = this.Get(result.ChannelId);
                        if (channel != null)
                        {
                            channel.OnSendComplete(e);
                        }

                        break;
                    }
                    default:
                        throw new ArgumentOutOfRangeException($"Tservice Update Error: {e.LastOperation}");
                }
            }
        }

        public override bool IsDisposed()
        {
            return this.acceptor == null;
        }

        public void Log()
        {
            Debug.ImportantLog($"当前真实链接数:{idChannels.Count}");
            foreach (var kv in idChannels)
            {
                kv.Value.Log();
            }
        }
    }
}