Unity Protobuf+RPC+UniTask

远程过程调用（RPC）协议详解

  • 什么是RPC协议
    • RPC的基本原理
    • RPC的关键组件
    • RPC的优缺点
    • Protobuf
    • 函数绑定
    • Call
    • Encode
    • Recv
    • Decode
    • Socket.Send和Recv
    • 项目地址

什么是RPC协议

远程过程调用（Remote Procedure Call，简称RPC）是一种网络通信协议，允许程序在不同的地址空间（通常在不同的物理计算机上）中调用彼此的方法，好像它们是在本地执行的一样。RPC隐藏了底层的网络通信细节，使开发人员能够像调用本地函数一样简单地调用远程服务。

RPC的基本原理

RPC的工作原理基于客户端-服务器模型，主要包括以下步骤：

1.客户端调用：客户端程序发起对某个远程过程的调用请求。
2.请求打包：调用参数被打包成消息，发送到服务器。
3.服务器解包和执行：服务器接收到消息，解包获取调用参数，执行相应的远程过程。
4.结果打包和返回：执行结果被打包成消息，发送回客户端。
5.客户端接收结果：客户端解包消息，获取调用结果

RPC的关键组件

1.客户端代理：负责将本地调用请求转换为远程调用请求，打包参数，并通过网络发送给服务器。
2.服务器代理：负责接收客户端的请求，解包参数，调用相应的服务方法，并将结果打包返回给客户端。
3.通信协议：定义客户端和服务器之间如何通信，常见的协议有HTTP、TCP等。
4.编解码器：负责参数和结果的序列化和反序列化，常见的格式有JSON、XML、Protobuf等。

RPC的优缺点

优点
1.简化远程调用：使得远程调用像本地调用一样简单，开发人员无需关心底层的网络通信细节。
2.语言无关：大多数RPC框架支持多种编程语言，方便不同语言的系统互操作。

缺点
1.调试困难：由于涉及网络通信，调试远程调用的问题比本地调用更加复杂。
2.可靠性要求高：需要处理网络延迟、丢包、超时等问题，增加了系统的复杂性。
3.耦合性：客户端和服务器需要共同遵循同一套接口定义，一旦接口发生变化，可能需要同时更新多个系统。

Protobuf

protoc.exe 生成C#文件
在这里插入图片描述

Gen.bat

@echo offrem 设置路径变量
set PROTOC_PATH="protoc.exe"
set PROTO_DIR="Protos"
set OUTPUT_DIR="ProtocolCodes"rem 创建日志头
echo .......................proto2C#.......................
echo.rem 检查目录是否存在
if not exist %PROTO_DIR% (echo Error: Protocols directory does not exist.echo Please create the Protocols directory and place your .proto files in it.echo.pauseexit /b
)rem 创建输出目录
if not exist %OUTPUT_DIR% mkdir %OUTPUT_DIR%rem 批量处理 .proto 文件
for %%f in (%PROTO_DIR%\*.proto) do (echo %%f complete%PROTOC_PATH% --proto_path=%PROTO_DIR% --csharp_out=%OUTPUT_DIR% %%f
)echo code generation complete. Press any key to close.
pause > nul

函数绑定

1.使用反射自动获取所有RPC函数, 对其进行Hash绑定

函数的定义 RPCMsgHandles.cs

public sealed class RPCMsgHandles
{private static void ReqMove(int unitId, Move move){}private static void RecvAttack(int skillid, Attack attack, ItemList itemList){LogHelper.Log($"Recv: skillid = {attack.Id}, targetId = {attack.TargetId}, itemList.Count = {itemList.Items.Count}");}private static void RecvDelete(int msg){LogHelper.Log($"Recv: state = {msg}");}private static void RecvReflectMove(Move move){LogHelper.Log($"move reflect sync: x:{move.X}, y:{move.Y}, speed:{move.Speed}, dir:{move.Dir}");}
}

使用反射进行函数绑定 RPCMoudle.cs

public sealed class RPCMoudle
{private static Dictionary<int, IRPC> _msg = new Dictionary<int, IRPC>();public static void Init(){System.Type type = typeof(RPCMsgHandles);MethodInfo[] methods = type.GetMethods(BindingFlags.Static | BindingFlags.NonPublic);foreach (MethodInfo methodInfo in methods){RPC method = new RPC(methodInfo);int index = 0;ParameterInfo[] infos = methodInfo.GetParameters();foreach (var info in infos){if (typeof(IMessage).IsAssignableFrom(info.ParameterType)){IMessage message = Activator.CreateInstance(info.ParameterType) as IMessage;method.AddParamType(DateType.Message);method.AddParam(index, message);}else{DateType dateType = GetDateType(info.ParameterType);method.AddParamType(dateType);}index++;}int hash = Globals.Hash(methodInfo.Name);if (_msg.ContainsKey(hash))throw new Exception("AddParamType rpc _method hash conflict: " + methodInfo.Name);_msg.Add(hash, method);}}
}

2.使用泛型手动进行RPC函数绑定

泛型类进行函数绑定 RPCMoudle.cs

public static void Register<T>(string methodName, Action<T> action) where T : class, IMessage, new()
{int id = Globals.Hash(methodName);RPCStatic<T> method = new RPCStatic<T>();method.Register(action, new T());if (_msg.ContainsKey(id)){LogHelper.LogError($"repeat id, id = {id}");}_msg[id] = method;
}public static void Unregister(string methodName)
{int id = Globals.Hash(methodName);if (_msg.ContainsKey(id)){_msg.Remove(id);}else{LogHelper.LogError($"no find method, id = {id}");}
}

Call

Call的实现, encode数据到byte[]，第一个参数必须为远程函数名字, 用于将函数名字的hashid写入数据头中, 这样远程服务器在解析数据的时候会先解析4字节的数据头表示函数的hashid

Call中的Send函数 是Socket发送协议, Send函数中会在数据头中写入数据的长度, 在接收方根据数据的长度接收完整数据 防止粘包

Call函数有多个方法重载, 根据业务需求使用
1.public static void Call(string methodName, IMessage message) 类型安全, 类型固定
2.public static void Call(string id, params object[] args) 类型不安全, 可以传入任何参数, 使用更加方便快捷

具体调用例子
Move move = new Move();
move.X = 10;
move.Y = 20;
move.Speed = 100;
move.Dir = 20;这里使用的是object[] args 类型不安全, 也会有装箱拆箱的开销, 使用这用方式需要前后端统一类型
使用起来简单方便, 业务逻辑开发上使用较为方便
比如请求领取奖励 RPCMoudle.Call("ReqAward", 传入表奖励id);
比如请求保存勾选 RPCMoudle.Call("Save", true);
RPCMoudle.Call("ReqMove", 10016, move);这样是类型安全的, 也不会存在装箱拆箱的开销
更加高效, 战斗场景较为适合
RPCMoudle.Call("ReqMove", move);

Call的实现, 将数据进行Encode转换成二进制

public static void Call(string methodName, IMessage message)
{if (message == null) return;try{int id = Globals.Hash(methodName);int offset = 0;BuffMessage msg = GameFrame.message.GetBuffMessage();BitConverter.TryWriteBytes(msg.bytes.AsSpan(offset), id);offset += sizeof(int);BitConverterHelper.WriteString(msg.bytes, ref offset, GameFrame.myRole.Id);BitConverterHelper.WriteMessage(msg.bytes, ref offset, message);msg.length = offset;Main.Instance.Send(msg);}catch(Exception ex){LogHelper.LogError(ex.ToString());}
}public static void Call(string id, params object[] args)
{try{Profiler.BeginSample("rpc call");int hash = Globals.Hash(id);BuffMessage msg = Encode(hash, args);Main.Instance.Send(msg);Profiler.EndSample();}catch(Exception ex){LogHelper.LogError(ex.ToString());}
}

Encode

Encode函数的实现

private static BuffMessage Encode(int id, params object[] args)
{int offset = 0;BuffMessage msg = GameFrame.message.GetBuffMessage();BitConverter.TryWriteBytes(msg.bytes.AsSpan(offset), id);offset += sizeof(int);BitConverterHelper.WriteString(msg.bytes, ref offset, GameFrame.myRole.Id);foreach (object arg in args){try{System.Type type = arg.GetType();switch (arg){case IMessage:BitConverterHelper.WriteMessage(msg.bytes, ref offset, (IMessage)arg);break;case Int16:BitConverterHelper.WriteInt16(msg.bytes, ref offset, (Int16)arg);break;case Int32:BitConverterHelper.WriteInt32(msg.bytes, ref offset, (Int32)arg);break;case Int64:BitConverterHelper.WriteInt64(msg.bytes, ref offset, (Int64)arg);break;case UInt16:BitConverterHelper.WriteUInt16(msg.bytes, ref offset, (UInt16)arg);break;case UInt32:BitConverterHelper.WriteUInt32(msg.bytes, ref offset, (UInt32)arg);break;case UInt64:BitConverterHelper.WriteUInt64(msg.bytes, ref offset, (UInt64)arg);break;case bool:BitConverterHelper.WriteBool(msg.bytes, ref offset, (bool)arg);break;case Byte:BitConverterHelper.WriteByte(msg.bytes, ref offset, (byte)arg);break;case SByte:BitConverterHelper.WriteByte(msg.bytes, ref offset, (byte)arg);break;case Char:BitConverterHelper.WriteChar(msg.bytes, ref offset, (Char)arg);break;case Single:BitConverterHelper.WriteSingle(msg.bytes, ref offset, (Single)arg);break;case Double:BitConverterHelper.WriteDouble(msg.bytes, ref offset, (Double)arg);break;case string:BitConverterHelper.WriteString(msg.bytes, ref offset, (string)arg);break;}}catch(Exception ex){LogHelper.LogError($"id: {id}, " + ex.ToString());msg.Dispose();return msg;}}msg.length = offset;return msg;
}

0GC的TryWriteBytes方案

namespace Game
{public static class BitConverterHelper{private static readonly int BUFFER_SIZE = 1024 * 1024;private static readonly byte[] buffer = new byte[BUFFER_SIZE];private static CodedOutputStream _stream;private static Stopwatch _watch;public static void Init(){CreateStream();_watch = new Stopwatch();_watch.Start();}private static void CreateStream(){if (_stream != null)_stream.Dispose();if (_watch != null){_watch.Stop();LogHelper.LogWarning($"create stream interval time: {_watch.ElapsedMilliseconds / 1000.0f} s");_watch.Restart();}_stream = new CodedOutputStream(buffer);}private static Span<byte> ToByteArray(IMessage message){if (message == null)return new byte[0];int length = message.CalculateSize();if (length == 0)return new byte[0];if (length >= BUFFER_SIZE){throw new Exception($"overflow: message length >= {BUFFER_SIZE}");}if (_stream.Position + length >= BUFFER_SIZE)CreateStream();int position = (int)_stream.Position;message.WriteTo(_stream);return buffer.AsSpan(position, length);}public static void WriteInt16(byte[] buffer, ref int offset, Int16 arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Int16;Check(buffer, offset + sizeof(Int16));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(Int16);}public static void WriteInt32(byte[] buffer, ref int offset, Int32 arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Int32;Check(buffer, offset + sizeof(Int32));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(Int32);}public static void WriteInt64(byte[] buffer, ref int offset, Int64 arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Int64;Check(buffer, offset + sizeof(Int64));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(Int64);}public static void WriteUInt16(byte[] buffer, ref int offset, UInt16 arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.UInt16;Check(buffer, offset + sizeof(UInt16));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(UInt16);}public static void WriteUInt32(byte[] buffer, ref int offset, UInt32 arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.UInt32;Check(buffer, offset + sizeof(UInt32));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(UInt32);}public static void WriteUInt64(byte[] buffer, ref int offset, UInt64 arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.UInt64;Check(buffer, offset + sizeof(UInt64));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(UInt64);}public static void WriteBool(byte[] buffer, ref int offset, bool arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Boolean;Check(buffer, offset + sizeof(bool));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(bool);}public static void WriteByte(byte[] buffer, ref int offset, byte arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Byte;Check(buffer, offset + 1);buffer[offset++] = arg;}public static void WriteChar(byte[] buffer, ref int offset, Char arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Char;Check(buffer, offset + sizeof(Char));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(Char);}public static void WriteSingle(byte[] buffer, ref int offset, Single arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Single;Check(buffer, offset + sizeof(Single));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(Single);}public static void WriteDouble(byte[] buffer, ref int offset, Double arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Double;Check(buffer, offset + sizeof(Double));BitConverter.TryWriteBytes(buffer.AsSpan(offset), arg);offset += sizeof(Double);}public static void WriteString(byte[] buffer, ref int offset, string arg){Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.String;byte[] bytes = Encoding.UTF8.GetBytes(arg);Check(buffer, offset + bytes.Length);BitConverter.TryWriteBytes(buffer.AsSpan(offset), bytes.Length);offset += sizeof(int);Span<byte> target = new Span<byte>(buffer, offset, buffer.Length - offset);bytes.CopyTo(target);offset += bytes.Length;}public static void WriteMessage(byte[] buffer, ref int offset, IMessage arg){IMessage message = arg;Span<byte> bytes = ToByteArray(message);Check(buffer, offset + 1);buffer[offset++] = (byte)DateType.Message;Check(buffer, offset + bytes.Length);BitConverter.TryWriteBytes(buffer.AsSpan(offset), bytes.Length);offset += sizeof(int);Span<byte> target = new Span<byte>(buffer, offset, bytes.Length);bytes.CopyTo(target);offset += bytes.Length;}private static void Check(byte[] buffer, int offset){if (offset >= buffer.Length)throw new Exception($"date length: {offset} > {Globals.DATA_SZIE}, Invalid data!!");}public static void Dispose(){_stream?.Dispose();_stream = null;}}
}

Recv

Decode 数据解析，调用本地方法


public static void OnRPC(BuffMessage msg)
{if(msg == null){LogHelper.LogError("socket recv error, msg == null");return;}Decode(msg.bytes);
}

Decode

0GC的Decode方案

private static void Decode(byte[] buffer)
{if (buffer == null || buffer.Length < sizeof(int)){LogHelper.LogError("Invalid buffer received");return;}int protoId = BitConverter.ToInt32(buffer, 0);if (!_msg.TryGetValue(protoId, out IRPC method)){LogHelper.LogError($"Method not found for protoId: {protoId}");return;}BuffMessage buffMessage = GameFrame.message.GetBuffMessage();try{Array.Copy(buffer, sizeof(int), buffMessage.bytes, 0, buffer.Length - sizeof(int));method.Decode(buffMessage.bytes);}catch (Exception ex){LogHelper.LogError($"Error invoking method for protoId {protoId}: {ex.Message}");}finally{GameFrame.message.PutBuffMessage(buffMessage);}
}

namespace Game
{public interface IRPC : IDisposable{public void Decode(byte[] buffer);}public abstract class RPCBase : IRPC{protected byte[] buffer;public abstract void Decode(byte[] buffer);protected ReadOnlySpan<byte> ReadData(DateType type, ref int offset){ReadOnlySpan<byte> data = null;int length = GetLength(type);if (length > 0){data = new ReadOnlySpan<byte>(buffer, offset, length);offset += length;}return data;}protected bool ToBoolean(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.Boolean, ref offset);return BitConverter.ToBoolean(data);}protected Byte ToByte(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.Char, ref offset);return data[0];}protected char ToChar(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.Char, ref offset);return BitConverter.ToChar(data);}protected Int16 ToInt16(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.Int16, ref offset);return BitConverter.ToInt16(data);}protected UInt16 ToUInt16(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.UInt16, ref offset);return BitConverter.ToUInt16(data);}protected Int32 ToInt32(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.Int32, ref offset);return BitConverter.ToInt32(data);}protected UInt32 ToUInt32(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.UInt32, ref offset);return BitConverter.ToUInt32(data);}protected Int64 ToInt64(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.Int64, ref offset);return BitConverter.ToInt64(data);}protected UInt64 ToUInt64(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.UInt64, ref offset);return BitConverter.ToUInt64(data);}protected Single ToSingle(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.Single, ref offset);return BitConverter.ToSingle(data);}protected Double ToDouble(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.Double, ref offset);return BitConverter.ToDouble(data);}protected string ToString(ref int offset){ReadOnlySpan<byte> data = ReadData(DateType.String, ref offset);return DecodeString(ref offset);}protected IMessage ToMessage(ref int offset, IMessage message){ReadOnlySpan<byte> data = ReadData(DateType.Message, ref offset);return DecodeMessage(ref offset, message);}private IMessage DecodeMessage(ref int offset, IMessage message){int length = BitConverter.ToInt32(buffer, offset);offset += sizeof(int);ReadOnlySpan<byte> messageData = new ReadOnlySpan<byte>(buffer, offset, length);offset += length;return message.Descriptor.Parser.ParseFrom(messageData)!;}private string DecodeString(ref int offset){int length = BitConverter.ToInt32(buffer, offset);offset += sizeof(int);ReadOnlySpan<byte> messageData = new ReadOnlySpan<byte>(buffer, offset, length);offset += length;return Encoding.UTF8.GetString(messageData);}private static int GetLength(DateType type){switch (type){case DateType.Boolean:return sizeof(bool);case DateType.Char:return sizeof(char);case DateType.SByte:case DateType.Byte:return sizeof(byte);case DateType.Int16:return sizeof(Int16);case DateType.UInt16:return sizeof(UInt16);case DateType.Int32:return sizeof(Int32);case DateType.UInt32:return sizeof(UInt32);case DateType.Int64:return sizeof(Int64);case DateType.UInt64:return sizeof(UInt64);case DateType.Single:return sizeof(Single);case DateType.Double:return sizeof(double);}return -1;}public virtual void Dispose(){buffer = null;}}
}

Decode数据到对象列表, 然后Invoke

namespace Game
{public class RPC : RPCBase{private MethodInfo _method;private List<DateType> _types;private List<object> _params;private Dictionary<int, IMessage> _param;private int _paramIndex;public RPC(MethodInfo method){this._method = method;_types = new List<DateType>();_params = new List<object>();_param = new Dictionary<int, IMessage>();}public void AddParamType(DateType type){_types?.Add(type!);}public void AddParam(int index, IMessage message){_param[index] = message;}public override void Decode(byte[] buffer){base.buffer = buffer;_paramIndex = 0;int offset = 0;_params.Clear();foreach (DateType type in _types){DateType dateType = (DateType)buffer[offset++];if (dateType != type){LogHelper.LogError($"dateType bo equals, recv: {Enum.GetName(typeof(DateType), type)} != local: {Enum.GetName(typeof(DateType), dateType)}");}object obj = ToObject(dateType, ref offset);_params.Add(obj!);_paramIndex++;}_method?.Invoke(null, _params.ToArray());}private object ToObject(DateType type, ref int offset){switch (type){case DateType.Message:IMessage message = null;if (!_param!.TryGetValue(_paramIndex, out message)){LogHelper.LogError("no find message");return null;}return ToMessage(ref offset, message);case DateType.Boolean:return ToBoolean(ref offset);case DateType.Char:return ToChar(ref offset);case DateType.SByte:case DateType.Byte:return ToByte(ref offset);case DateType.Int16:return ToInt16(ref offset);case DateType.UInt16:return ToUInt16(ref offset);case DateType.Int32:return ToInt32(ref offset);case DateType.UInt32:return ToUInt32(ref offset);case DateType.Int64:return ToInt64(ref offset);case DateType.UInt64:return ToUInt64(ref offset);case DateType.Single:return ToSingle(ref offset);case DateType.Double:return ToDouble(ref offset);case DateType.String:return ToString(ref offset);default:LogHelper.LogError("no find dateType: " + type);break;}return null;}public override void Dispose(){base.Dispose();_method = null;_types = null;}}
}

泛型Decode，然后Invokde

namespace Game
{public class RPCStatic<T> : RPCBase{private Action<T> _action;private IMessage _message;public RPCStatic(){}public virtual void Register(Action<T> action, IMessage message){this._message = message;this._action = action;}public override void Decode(byte[] buffer){base.buffer = buffer;int offset = 0;DateType dateType = (DateType)buffer[offset++];try{if (dateType == DateType.Message){IMessage arg = ToMessage(ref offset, _message);_action?.Invoke((T)arg);}else{LogHelper.LogError($"invoke error, type != DateType.Message, type = {dateType}");}}catch (Exception ex){LogHelper.LogError(ex.ToString());}}public override void Dispose(){}}
}

Socket.Send和Recv

使用UniTask实现的多线程异步收发消息，处理了超时重发和异常处理，接收消息时的粘包处理

namespace Game
{public enum SocketState{None = 0,Connected = 1,Disconnected = 2,Connecting = 3,ConnectFailed = 4,Close = 5,Dispose = 6,}public class Tcp{private ConcurrentQueue<BuffMessage> _sendMsgs;private ConcurrentQueue<BuffMessage> _receiveMsgs;private TcpClient _tcpClient;private SocketState _socketState;private byte[] _recvBuff;private int _recvOffset;private int _delay = 10;private CancellationTokenSource _recvCancelToken;private CancellationTokenSource _sendCancelToken;public SocketState State { get { return _socketState; } }public string IP { get; set; }public int Port { get; set; }public NetworkStream Stream{get { return _tcpClient.GetStream(); }}public Tcp(){_sendMsgs = new ConcurrentQueue<BuffMessage>();_receiveMsgs = new ConcurrentQueue<BuffMessage>();_recvBuff = new byte[Globals.BUFFER_SIZE];}private void InitTcpClient(){_tcpClient = new TcpClient();_recvCancelToken = new CancellationTokenSource();_sendCancelToken = new CancellationTokenSource();}public void Update(){Profiler.BeginSample("on tcp rpc");if (_receiveMsgs.TryDequeue(out BuffMessage msg)){RPCMoudle.OnRPC(msg);GameFrame.message.PutBuffMessage(msg);}Profiler.EndSample();}public void Connect(string ip, int port){IP = ip;Port = port;Connect();}public async void Connect(){try{Close();InitTcpClient();SetSocketState(SocketState.Connecting);await _tcpClient.ConnectAsync(IP, Port);OnConnect();}catch (Exception ex){LogHelper.LogError(ex.ToString());}}private void OnConnect(){try{if (_tcpClient.Connected){LogHelper.Log("connected...");SetSocketState(SocketState.Connected);StartAsyncTasks();}else{SetSocketState(SocketState.ConnectFailed);}}catch (Exception ex){LogHelper.LogError("连接或通信发生错误：{0}" + ex.Message);SetSocketState(SocketState.ConnectFailed);}}private void StartAsyncTasks(){UniTask send = UniTask.Create(SendThread);UniTask recv = UniTask.Create(RecvThread);}private async UniTask SendThread(){await UniTask.SwitchToThreadPool();while (_socketState == SocketState.Connected){while (true){if (!_sendMsgs.TryDequeue(out BuffMessage msg))break;var timeoutToken = new CancellationTokenSource();timeoutToken.CancelAfterSlim(TimeSpan.FromMilliseconds(msg.TimeoutMillisecond));var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(_sendCancelToken.Token, timeoutToken.Token);try{if(_sendCancelToken.IsCancellationRequested)break;await Stream.WriteAsync(msg.bytes, 0, msg.length, linkedCts.Token);LogHelper.Log($"发送完成: {msg.length} byte");GameFrame.message.PutBuffMessage(msg);}catch (OperationCanceledException ex){if (timeoutToken.IsCancellationRequested){_sendMsgs.Enqueue(msg);LogHelper.LogWarning("消息发送超时, 添加到队列末尾, 等待发送...");await UniTask.Delay(10);continue;}LogHelper.LogWarning("发送操作被终止..." + ex.Message);break;}catch (IOException ex) when (ex.InnerException is SocketException socketEx && socketEx.SocketErrorCode == SocketError.ConnectionAborted){LogHelper.Log("发送操作被终止...");break;}catch (Exception ex){LogHelper.LogError("发送错误: " + ex.Message);break;}}await UniTask.Delay(_delay);}}private async UniTask RecvThread(){await UniTask.SwitchToThreadPool();while (_socketState == SocketState.Connected){try{if (_recvCancelToken.IsCancellationRequested) break;int length = await Stream.ReadAsync(_recvBuff, _recvOffset, _recvBuff.Length - _recvOffset, _recvCancelToken.Token);if (length == 0){LogHelper.Log("connect failed...");break;}_recvOffset += length;int offset = 0;while (true){if (_recvOffset - offset < sizeof(int))// 没有足够的数据读取下一个消息的长度break;int dataLength = BitConverter.ToInt32(_recvBuff, offset);if (_recvOffset - offset < dataLength + sizeof(int))// 没有足够的数据读取完整的消息break;// 读取完整消息BuffMessage msg = GameFrame.message.GetBuffMessage();Buffer.BlockCopy(_recvBuff, offset + sizeof(int), msg.bytes, 0, dataLength);_receiveMsgs.Enqueue(msg);// 移动偏移量到下一个消息offset += sizeof(int) + dataLength;}// 将未处理的数据移到缓冲区开头if (_recvOffset - offset > 0)Buffer.BlockCopy(_recvBuff, offset, _recvBuff, 0, _recvOffset - offset);_recvOffset -= offset;}catch(OperationCanceledException ex){LogHelper.Log("读取操作被终止: " + ex.Message);break;}catch (IOException ex) when (ex.InnerException is SocketException socketEx && socketEx.SocketErrorCode == SocketError.OperationAborted){LogHelper.Log("读取操作被终止...");break;}catch (Exception ex){LogHelper.LogError("读取错误: " + ex.ToString());break;}await UniTask.Delay(_delay);}}private void SetSocketState(SocketState state){_socketState = state;}public void Send(BuffMessage message){if (message.length > 0){int headLength = sizeof(int);Buffer.BlockCopy(message.bytes, 0, message.bytes, headLength, message.length);BitConverter.TryWriteBytes(message.bytes.AsSpan(0), message.length);message.length += headLength;_sendMsgs.Enqueue(message);}else{GameFrame.message.PutBuffMessage(message);}}public void Close(){if (_tcpClient == null)return;try{if (_tcpClient.Connected){SetSocketState(SocketState.Close);_recvCancelToken.Dispose();_sendCancelToken.Dispose();_tcpClient.Close();}}catch (Exception ex){LogHelper.LogError(ex.ToString());}}public void Dispose(){Close();if (_tcpClient != null){_tcpClient.Dispose();_tcpClient = null;}if (_sendMsgs != null){_sendMsgs.Clear();_sendMsgs = null;}if (_receiveMsgs != null){_receiveMsgs.Clear();_receiveMsgs = null;}SetSocketState(SocketState.Dispose);}}
}

项目地址

SimpleRPC

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.xdnf.cn/news/1450993.html

如若内容造成侵权/违法违规/事实不符，请联系一条长河网进行投诉反馈，一经查实，立即删除！

相关文章

自动化测试 —— ReadyAPI赋能API性能测试，助力应对高峰期流量挑战！

自动化测试 —— ReadyAPI赋能API性能测试，助力应对高峰期流量挑战！

在当今数字驱动的市场中&#xff0c;API的完美性能对于企业在高峰期提升营业收入至关重要。随着消费者越来越依赖于在线购物和移动App购物&#xff0c;任何与API相关的故障或减速都可能导致顾客体验变差和交易流失&#xff0c;从而造成销售损失。因此&#xff0c;企业需要优先考…
阅读更多...
EasyExcel文件导出，出现有文件但没有数据的问题

EasyExcel文件导出，出现有文件但没有数据的问题

一开始由于JDK版本过高&#xff0c;我用的17&#xff0c;一直excel没有数据&#xff0c;表头也没有&#xff0c;后来摸索了好久&#xff0c;找了资料也没有&#xff0c;后来改了代码后报了一个错误&#xff08;com.alibaba.excel.exception.ExcelGenerateException: java.lang.…
阅读更多...
计算机网络 —— 应用层（应用层概述及服务方式）

计算机网络 —— 应用层（应用层概述及服务方式）

计算机网络 —— 应用层&#xff08;应用层概述及服务方式&#xff09; 应用层服务方式C/S&#xff08;客户端-服务器&#xff08;C/S&#xff09;模型&#xff09;基本概念特点B/S&#xff08;Browser/Server&#xff09;基本概念特点应用场景 p2p &#xff08;对等网络&#…
阅读更多...
定时清理rocketmq日志--crontab

定时清理rocketmq日志--crontab

1、背景 之前在部署rocketmq的时候未修改日志路径&#xff0c;导致在用户目录下有日志数据写入。因不方便修改或空间足够可正常写入&#xff0c;但日志量过大需清理&#xff0c;现添加定时任务执行。 2、规划&#xff1a; 目前测试阶段&#xff0c;所以时间是可变的&#xf…
阅读更多...
数据预处理之基于聚类的TOD异常值检测#matlab

数据预处理之基于聚类的TOD异常值检测#matlab

1.基于聚类的异常值检测方法 物以类聚——相似的对象聚合在一起&#xff0c;基于聚类的异常点检测方法有两个共同特点&#xff1a; (1)先采用特殊的聚类算法处理输入数据而得到聚类&#xff0c;再在聚类的基础上来检测异常。 (2)只需要扫描数据集若干次&#xff0c;效率较高…
阅读更多...
MATLAB R2023a for Mac(商业数学软件)中文激活版

MATLAB R2023a for Mac(商业数学软件)中文激活版

MATLAB R2023a for Mac 是MathWorks打造的一款非常专业且强大的商业数学软件&#xff0c;MATLAB允许矩阵操作&#xff0c;函数和数据绘图&#xff0c;算法实现&#xff0c;用户界面创建&#xff0c;以及与其他语言&#xff08;包括C&#xff0c;C &#xff0c;C&#xff03;&am…
阅读更多...
蚂蚁分类信息系统二开仿么么街货源客模板微商货源网源码(带手机版)

蚂蚁分类信息系统二开仿么么街货源客模板微商货源网源码(带手机版)

源码介绍 网站采用蚂蚁分类信息系统二次开发&#xff0c;模板仿么么街货源客模板&#xff0c;微商货源网定制版。 模板设计风格简洁&#xff0c;分类信息采用列表形式发布&#xff0c;这种设计方式非常符合度娘 SEO 规则。收录效果是杠杠的。 这个网站风格目前是用来做货源推…
阅读更多...
同城如何异地共享文件？

同城如何异地共享文件？

在现代社会中&#xff0c;跨地区的合作变得越来越普遍&#xff0c;而这也带来了共享文件的需求。当我们身处不同的城市&#xff0c;如何高效地共享文件已经成为一项迫切的需求。本文将介绍一种名为“同城异地共享文件”的解决方案&#xff0c;帮助解决这一问题。 2. 天联组网—…
阅读更多...
【elementui源码解析】如何实现自动渲染md文档-第三篇

【elementui源码解析】如何实现自动渲染md文档-第三篇

目录 1.前言 2.webpack.demo.js 3.markdown文档 4.fence.js 1&#xff09;tokens 2&#xff09;::: 3&#xff09; 5.containers.js 1&#xff09;markdown-it-container 2&#xff09;md.use() 3&#xff09;代码逻辑 4&#xff09;containers小结 6.congfig.js …
阅读更多...
VMware ESXi 8.0U2c macOS Unlocker OEM BIOS ConnectX-3 网卡定制版 (集成驱动版)

VMware ESXi 8.0U2c macOS Unlocker OEM BIOS ConnectX-3 网卡定制版 (集成驱动版)

VMware ESXi 8.0U2c macOS Unlocker & OEM BIOS ConnectX-3 网卡定制版 (集成驱动版) 发布 ESXi 8.0U2 集成驱动版&#xff0c;在个人电脑上运行企业级工作负载 请访问原文链接&#xff1a;https://sysin.org/blog/vmware-esxi-8-u2-sysin/&#xff0c;查看最新版。原创作…
阅读更多...
碎片化知识如何被系统性地吸收？

碎片化知识如何被系统性地吸收？

一、方法论 碎片化知识指的是通过各种渠道快速获取的零散信息和知识点&#xff0c;这些信息由于其不完整性和孤立性&#xff0c;不易于记忆和应用。为了系统性地吸收碎片化知识&#xff0c;可以采用以下策略&#xff1a; 1. **构建知识框架**&#xff1a; - 在开始吸收之前&am…
阅读更多...
3DS MAX课程结业实验报告书

3DS MAX课程结业实验报告书

3DS MAX课程结业实验报告书 目录 3DS MAX课程结业实验报告书 n 一、实验内容 n 二、实验说明 n 三、实验过程 n 3.1 安装插件 n 3.2 实验过程 1.了解常用操作 2.了解PolyCloth 的使用 3.新建沙发底 在加两个扶翼和地板砖 4.时钟 5.初级沙发 6.高级沙…
阅读更多...
【python】tkinter编程三大布局管理器pack、grid、place应用实战解析

【python】tkinter编程三大布局管理器pack、grid、place应用实战解析

✨✨ 欢迎大家来到景天科技苑✨✨ &#x1f388;&#x1f388; 养成好习惯&#xff0c;先赞后看哦~&#x1f388;&#x1f388; &#x1f3c6; 作者简介&#xff1a;景天科技苑 &#x1f3c6;《头衔》&#xff1a;大厂架构师&#xff0c;华为云开发者社区专家博主&#xff0c;…
阅读更多...
RAG_Example

RAG_Example

今天尝试基于langchain进行LLM RAG搭建&#xff0c;感觉使用难度没有想象中大。具体流程参考末尾链接。 主要流程包括下面几个模块&#xff0c;每一个模块都有很多选择&#xff0c;而不是唯一解。 但这里可以感受到潜在的几个问题 1. 文本转换过程中&#xff0c;PDF的信息可…
阅读更多...
Apache Doris单机快速安装（已踩坑）

Apache Doris单机快速安装（已踩坑）

官方文档&#xff1a;https://doris.incubator.apache.org/zh-CN/docs/get-starting/quick-start/ 环境&#xff1a; 操作系统&#xff1a;CentOS7.6 X86_64 JDK&#xff1a;Oracle jdk1.8.0_351 1.版本下载 从 doris.apache.org 下载相应的 Doris 安装包&#xff0c;并且解压…
阅读更多...
多表联查小情景例子

多表联查小情景例子

这里有个需求&#xff1a; 需要给定套餐表setmeal 的 id 查询这个套餐内的所有菜品 dish 实际上如果可能&#xff0c;只需要查询套餐菜品表(setmeal_dish)&#xff0c;查看一个套餐内包含包含的所有菜品&#xff0c;然后把这些菜品全部取出即可。 看一下setmeal_dish 表&…
阅读更多...
【教程】从0开始搭建大语言模型：构造GPT模型

【教程】从0开始搭建大语言模型：构造GPT模型

从0开始搭建大语言模型&#xff1a;构造GPT模型 从0开始搭建大语言模型&#xff1a;构造GPT模型GPT模型Layer NormalizationGELU激活函数Feed Forward网络增强shortcut连接构造Transformer Block构造GPT模型使用GPT模型生成文本 从0开始搭建大语言模型&#xff1a;构造GPT模型 …
阅读更多...
maven archetype项目构架

maven archetype项目构架

1、设置环境变量 set MAVEN_HOMED:\SF\java\apache-maven-3.6.3 set path%path%;%MAVEN_HOME%\bin;2、制作archetype mvn -s "D:\SF\java\apache-maven-3.6.3\conf\settings.xml" archetype:create-from-project -DpackageNamecom.demo.esb-s:指定maven的setting文…
阅读更多...
【Hudi】核心概念-基本概念

【Hudi】核心概念-基本概念

目录 基本概念时间轴(TimeLine)Instant action:在表上执行的操作类型Instant timeState两个时间概念 文件布局(File Layout)索引(Index)原理索引选项全局索引与非全局索引索引的选择策略 表类型(Table Types)查询类型Snapshot QueriesIncremental QueriesRead Optimized Querie…
阅读更多...
图像生成新篇章：Stable Diffusion 3 Medium开源评析

图像生成新篇章：Stable Diffusion 3 Medium开源评析

摘要 在数字艺术与人工智能的交汇点上&#xff0c;Stable Diffusion 3&#xff08;SD3&#xff09;的开源无疑是一场技术革新的盛宴。就在3月份&#xff0c;我撰写了一篇博文&#xff0c;深入探讨了SD3的技术报告内容与介绍&#xff0c;文章发表在CSDN博客上&#xff0c;https:…
阅读更多...
最新文章

Copyright @ 2022~2023