4.消息池类
4.1 知识点
创建消息池类
- 消息池类：主要用于注册消息类型、消息处理器类型与ID的映射关系，以便在运行时根据ID获取相应的消息类对象或消息处理类对象。
public class MessagePool
{
private Dictionary<int, Type> messsages = new Dictionary<int, Type>();
private Dictionary<int, Type> handlers = new Dictionary<int, Type>();
public MessagePool()
{
Register(1001, typeof(PlayerMessage), typeof(PlayerMessageHandler));
}
private void Register(int id, Type messageType, Type handlerType)
{
messsages.Add(id, messageType);
handlers.Add(id, handlerType);
}
/// <summary>
/// 根据ID获取指定的消息类对象
/// </summary>
/// <param name="id"></param>
/// <returns></returns>
public BaseMessage GetMessage(int id)
{
if (!messsages.ContainsKey(id))
return null;
return Activator.CreateInstance(messsages[id]) as BaseMessage;
}
/// <summary>
/// 根据ID获取指定的消息处理类对象
/// </summary>
/// <param name="id"></param>
/// <returns></returns>
public BaseHandler GetHandler(int id)
{
if (!handlers.ContainsKey(id))
return null;
return Activator.CreateInstance(handlers[id]) as BaseHandler;
}
}
在TCP异步管理器中定义消息池对象
- 用途：在 TCP 异步管理器中定义消息池对象，以便在分包黏包方法中使用消息池获取消息对象和消息处理器，并将它们压入队列中。
// 消息池对象，用于快速获取消息和消息处理类对象
private MessagePool messagePool = new MessagePool();
// 在分包黏包方法中使用消息池获取消息对象和消息处理器，并将它们压入队列中
BaseMessage baseMessage = messagePool.GetMessage(msgID);
if (baseMessage != null)
{
// 反序列化
baseMessage.Reading(cacheBytes, nowIndex);
// 获取消息处理器对象
BaseHandler baseHandler = messagePool.GetHandler(msgID);
baseHandler.message = baseMessage;
// 将消息处理器对象放入队列中，稍后在Update中进行处理
receiveQueue.Enqueue(baseHandler);
}
之后会写自动生成消息池和Handler类的工具
4.2 知识点代码
MessagePool
using GamePlayer;
using GameSystem;
using System;
using System.Collections;
using System.Collections.Generic;
using UnityEngine;
/// <summary>
/// 消息池类 主要是用于 注册 ID和消息类型以及消息处理器类型的映射关系
/// 方便我们获取对象 进行反序列化和消息逻辑处理
/// </summary>
public class MessagePool
{
// 记录消息类型和ID的映射关系
private Dictionary<int, Type> messsages = new Dictionary<int, Type>();
// 记录消息处理器类型和ID的映射关系
private Dictionary<int, Type> handlers = new Dictionary<int, Type>();
// 构造函数，用于初始化消息池并注册初始映射关系
public MessagePool()
{
// 在构造函数中进行注册映射关系
Register(1001, typeof(PlayerMessage), typeof(PlayerMessageHandler));
}
// 注册方法，将消息类型、消息处理器类型与ID进行关联
private void Register(int id, Type messageType, Type handlerType)
{
messsages.Add(id, messageType);
handlers.Add(id, handlerType);
}
/// <summary>
/// 根据ID得到一个指定的消息类对象
/// </summary>
/// <param name="id"></param>
/// <returns></returns>
public BaseMessage GetMessage(int id)
{
// 如果消息池中不包含指定ID的映射关系，则返回null
if (!messsages.ContainsKey(id))
return null;
// 根据ID创建对应消息类对象并返回
return Activator.CreateInstance(messsages[id]) as BaseMessage;
}
/// <summary>
/// 根据ID得到一个指定的消息处理类对象
/// </summary>
/// <param name="id"></param>
/// <returns></returns>
public BaseHandler GetHandler(int id)
{
// 如果消息池中不包含指定ID的映射关系，则返回null
if (!handlers.ContainsKey(id))
return null;
// 根据ID创建对应消息处理类对象并返回
return Activator.CreateInstance(handlers[id]) as BaseHandler;
}
}
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();
//消息池对象 用于快速获取消息和消息处理类对象
private MessagePool messagePool = new MessagePool();
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);
//得到一个指定ID的消息类对象 只不过是用父类装子类
BaseMessage baseMessage = messagePool.GetMessage(msgID);
if (baseMessage != null)
{
//反序列化
baseMessage.Reading(cacheBytes, nowIndex);
//得到一个消息处理器对象
BaseHandler baseHandler = messagePool.GetHandler(msgID);
baseHandler.message = baseMessage;
//把消息处理器对象 放入队列中 稍后在Update中进行处理
receiveQueue.Enqueue(baseHandler);
}
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
