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
