24.总结

3.Handler类

---

3.1 知识点

创建消息处理器抽象基类

• 消息处理器基类 主要用于处理消息的逻辑的

public abstract class BaseHandler
{
    //处理者处理哪个消息
    public BaseMessage message;

    //真正处理消息的方法
    public abstract void MessageHandle();
}

在TCP异步管理器中，修改消息队列

//private Queue<BaseMessage> receiveQueue = new Queue<BaseMessage>(); // 创建一个队列，用于存储接收到的消息
private Queue<BaseHandler> receiveQueue = new Queue<BaseHandler>(); // 创建一个队列，用于存储消息接受者

修改每一帧检查消息队列的逻辑

void Update()
{
    if (receiveQueue.Count > 0)
    {
        BaseHandler baseHandler = receiveQueue.Dequeue();
        baseHandler.MessageHandle();
        //BaseMessage baseMessage = receiveQueue.Dequeue();
        //switch (baseMessage)
        //{
        //    case PlayerMessage msg:
        //        print(msg.playerID);
        //        print(msg.playerData.name);
        //        print(msg.playerData.lev);
        //        print(msg.playerData.atk);
        //        break;
        //}
    }
}

实现玩家消息类处理器

public class PlayerMessageHandler : BaseHandler
{
    public override void MessageHandle()
    {
        PlayerMessage msg = message as PlayerMessage;
        //以后我们处理对应某一个消息的逻辑 只需要在消息处理者对象的
        //消息处理方法中 写逻辑就行了
        Debug.Log(msg.playerID);
    }
}

在处理分包黏包方法中，修改放入队列的对象

//解析消息体
BaseMessage baseMessage = null;

BaseHandler baseHandler = null;

switch (msgID)
{
    case 1001:
        baseMessage = new PlayerMessage();
        baseMessage.Reading(cacheBytes, nowIndex);

        baseHandler = new PlayerMessageHandler();
        baseHandler.message = baseMessage;
        break;
}
if (baseHandler != null)
//if (baseMessage != null)
    receiveQueue.Enqueue(baseHandler);
    //receiveQueue.Enqueue(baseMessage);

---

3.2 知识点代码

BaseHandler

using System.Collections;
using System.Collections.Generic;
using UnityEngine;

/// <summary>
/// 消息处理器基类 主要用于处理消息的逻辑的
/// </summary>
public abstract class BaseHandler
{
    //处理者处理哪个消息
    public BaseMessage message;

    //真正处理消息的方法
    public abstract void MessageHandle();
}

PlayerMessageHandler

using System.Collections;
using System.Collections.Generic;
using UnityEngine;

public class PlayerMessageHandler : BaseHandler
{
    public override void MessageHandle()
    {
        PlayerMessage msg = message as PlayerMessage;
        //以后我们处理对应某一个消息的逻辑 只需要在消息处理者对象的
        //消息处理方法中 写逻辑就行了
        Debug.Log(msg.playerID);
    }
}

TcpNetAsyncManager

using System;
using System.Collections;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Text;
using UnityEngine;

public class TcpNetAsyncManager : BaseSingletonInMonoBehaviour<TcpNetAsyncManager>
{
    //和服务器进行连接的 Socket
    private Socket socket;// 创建Socket对象，用于网络通信

    //private Queue<BaseMessage> receiveQueue = new Queue<BaseMessage>(); // 创建一个队列，用于存储接收到的消息
    private Queue<BaseHandler> receiveQueue = new Queue<BaseHandler>(); // 创建一个队列，用于存储接消息接受者

    //用于处理分包时 缓存的 字节数组 和 字节数组长度
    private byte[] cacheBytes = new byte[1024 * 1024];
    private int cacheNum = 0;

    //发送心跳消息的间隔时间
    private int SEND_HEART_MSG_TIME = 2;
    private HeartMessage heartMessage = new HeartMessage();

    protected override void Awake()
    {
        base.Awake();

        //客户端循环定时给服务端发送心跳消息
        InvokeRepeating("SendHeartMsg", 0, SEND_HEART_MSG_TIME);
    }

    //发送心跳消息
    private void SendHeartMsg()
    {
        if (socket != null && this.socket.Connected)
            Send(heartMessage);
    }

    //连接服务器的代码
    public void Connect(string ip, int port)
    {
        if (socket != null && socket.Connected)
            return;

        IPEndPoint ipPoint = new IPEndPoint(IPAddress.Parse(ip), port); // 创建一个IP终结点对象
        socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

        SocketAsyncEventArgs socketAsyncEventArgs = new SocketAsyncEventArgs();
        socketAsyncEventArgs.RemoteEndPoint = ipPoint;
        socketAsyncEventArgs.Completed += (socket, socketAsyncEventArgs) =>
        {
            if (socketAsyncEventArgs.SocketError == SocketError.Success)
            {
                print("连接成功");
                //收消息
                SocketAsyncEventArgs receiveSocketAsyncEventArgs = new SocketAsyncEventArgs();
                receiveSocketAsyncEventArgs.SetBuffer(cacheBytes, 0, cacheBytes.Length);
                receiveSocketAsyncEventArgs.Completed += ReceiveCallBack;
                this.socket.ReceiveAsync(receiveSocketAsyncEventArgs);
            }
            else
            {
                print("连接失败" + socketAsyncEventArgs.SocketError);
            }
        };
        socket.ConnectAsync(socketAsyncEventArgs);
    }

    // 关闭连接
    public void Close()
    {
        print("客户端主动断开连接");
        if (socket != null)
        {
            //主动发送一条断开的消息个服务端 
            QuitMessage quitMessage = new QuitMessage();

            //这里不能用我们封装的Send方法 因为Send方法是开一个线程发送的 可能还没发就直接被断开了
            socket.Send(quitMessage.Writing());

            socket.Shutdown(SocketShutdown.Both); // 关闭套接字的发送和接收

            socket.Disconnect(false);//手动停止连接 参数意思是这个socket以后还用不用

            socket.Close(); // 关闭套接字连接

            socket = null;//当前socket不会再用了 置空
        }
    }

    // 当对象被销毁时，确保关闭连接
    private void OnDestroy()
    {
        Close(); // 调用关闭连接的方法
    }

    /// <summary>
    /// 用于测试 直接发字节数组的方法
    /// </summary>
    /// <param name="bytes"></param>
    public void SendTest(byte[] bytes)
    {
        SocketAsyncEventArgs socketAsyncEventArgs = new SocketAsyncEventArgs();
        socketAsyncEventArgs.SetBuffer(bytes, 0, bytes.Length);
        socketAsyncEventArgs.Completed += (socket, args) =>
        {
            if (args.SocketError != SocketError.Success)
            {
                print("发送消息失败" + args.SocketError);
                Close();
            }

        };
        this.socket.SendAsync(socketAsyncEventArgs);
    }

    // 发送消息
    public void Send(BaseMessage baseMessage)
    {
        if (this.socket != null && this.socket.Connected)
        {
            byte[] bytes = baseMessage.Writing();
            SocketAsyncEventArgs args = new SocketAsyncEventArgs();
            args.SetBuffer(bytes, 0, bytes.Length);
            args.Completed += (socket, args) =>
            {
                if (args.SocketError != SocketError.Success)
                {
                    print("发送消息失败" + args.SocketError);
                    Close();
                }

            };
            this.socket.SendAsync(args);
        }
        else
        {
            Close();
        }
    }

    //收消息完成的回调函数
    private void ReceiveCallBack(object obj, SocketAsyncEventArgs socketAsyncEventArgs)
    {
        if (socketAsyncEventArgs.SocketError == SocketError.Success)
        {
            HandleReceiveMsg(socketAsyncEventArgs.BytesTransferred);
            //继续去收消息
            socketAsyncEventArgs.SetBuffer(cacheNum, socketAsyncEventArgs.Buffer.Length - cacheNum);
            //继续异步收消息
            if (this.socket != null && this.socket.Connected)
                socket.ReceiveAsync(socketAsyncEventArgs);
            else
                Close();
        }
        else
        {
            print("接受消息出错" + socketAsyncEventArgs.SocketError);
            //关闭客户端连接
            Close();
        }
    }

    //处理接受消息 分包、黏包问题的方法
    private void HandleReceiveMsg(int receiveNum)
    {
        int msgID = 0;
        int msgLength = 0;
        int nowIndex = 0;

        cacheNum += receiveNum;

        while (true)
        {
            //每次将长度设置为-1 是避免上一次解析的数据 影响这一次的判断
            msgLength = -1;
            //处理解析一条消息
            if (cacheNum - nowIndex >= 8)
            {
                //解析ID
                msgID = BitConverter.ToInt32(cacheBytes, nowIndex);
                nowIndex += 4;
                //解析长度
                msgLength = BitConverter.ToInt32(cacheBytes, nowIndex);
                nowIndex += 4;
            }

            if (cacheNum - nowIndex >= msgLength && msgLength != -1)
            {
                //解析消息体
                BaseMessage baseMessage = null;

                BaseHandler baseHandler = null;

                switch (msgID)
                {
                    case 1001:
                        baseMessage = new PlayerMessage();
                        baseMessage.Reading(cacheBytes, nowIndex);

                        baseHandler = new PlayerMessageHandler();
                        baseHandler.message = baseMessage;
                        break;
                }
                if (baseHandler != null)
                //if (baseMessage != null)
                    receiveQueue.Enqueue(baseHandler);
                    //receiveQueue.Enqueue(baseMessage);


                nowIndex += msgLength;
                if (nowIndex == cacheNum)
                {
                    cacheNum = 0;
                    break;
                }
            }
            else
            {
                if (msgLength != -1)
                    nowIndex -= 8;
                //就是把剩余没有解析的字节数组内容 移到前面来 用于缓存下次继续解析
                Array.Copy(cacheBytes, nowIndex, cacheBytes, 0, cacheNum - nowIndex);
                cacheNum = cacheNum - nowIndex;
                break;
            }
        }

    }

    void Update()
    {
        if (receiveQueue.Count > 0)
        {
            BaseHandler baseHandler = receiveQueue.Dequeue();
            baseHandler.MessageHandle();
            //BaseMessage baseMessage = receiveQueue.Dequeue();
            //switch (baseMessage)
            //{
            //    case PlayerMessage msg:
            //        print(msg.playerID);
            //        print(msg.playerData.name);
            //        print(msg.playerData.lev);
            //        print(msg.playerData.atk);
            //        break;
            //}
        }
    }

}

---

---

转载请注明来源，欢迎对文章中的引用来源进行考证，欢迎指出任何有错误或不够清晰的表达。可以在下面评论区评论，也可以邮件至 785293209@qq.com