ProtoBuff3 unity_TCP网络发包解包&&消息订阅

using Google.Protobuf;

//using Google.Protobuf.Examples.AddPerson;

using Google.Protobuf.WellKnownTypes;

using System;

using System.Net.Sockets;

using UnityEngine;

using ARProto;

using pb = global::Google.Protobuf;

public class NewBehaviourScript : MonoBehaviour

{

void OnGUI()

{

if (GUI.Button(new Rect(100, 10, 120, 100), new GUIContent("Button", "Go")))

{

UIGo();

}

}

public void UIGo()

{

       Person person =new   Person();

       person.Id=01;

       AddPerson addPerson =new AddPerson();

SendMsg(addPerson);

}

// Use this for initialization

void Start()

{

StartConnect();

}

TcpClient tcpClient; //

byte[] receive_buff; // 专门用来接收Socket里面的数据的

byte[] data_buff; // 用来存当前未处理的数据

CodedOutputStream outputStream; // 用来绑定SocketStream，方便把proto对象转换成字节流Stream输送给服务器

void StartConnect()

{

TcpClient client = new TcpClient();

tcpClient = client;

//这里写上你自己服务器的ip和端口

client.Connect("106.2.124.243", 9090);

receive_buff = new byte[client.ReceiveBufferSize];

outputStream = new CodedOutputStream(client.GetStream());

// 监听一波服务器消息

client.GetStream().BeginRead(receive_buff, 0, client.ReceiveBufferSize, ReceiveMessage, null);

}

int nFalg = 0; // 这个变量主要是为了防止和服务端无休无止互发消息，测试代码

void Update()

{

// 因为ReceiveMessage接收数据是异步的方式，不是在主线程，有些方法不能用，比如ToString，所以消息处理放在这里处理

// 但主要是因为后面要加上消息广播，可以添加在这里

if (data_buff != null && ++nFalg < 5)

{

// 把数据传给CodedInputStream计算本次包的长度

CodedInputStream inputStream = new CodedInputStream(data_buff);

int length = inputStream.ReadLength();

// 计算"包长度"占用的字节数，后面取数据的时候扣掉这个字节数，就是真实数据长度

int lengthLength = CodedOutputStream.ComputeLengthSize(length);

// 当前数据足够解析一个包了

if (length + lengthLength <= data_buff.Length)

{

byte[] real_data = new byte[length];

// 拷贝真实数据

Array.Copy(data_buff, lengthLength, real_data, 0, length);

// 假设服务器给你发了个AddressBook

AddPerson ab = AddPerson.Parser.ParseFrom(real_data);

// 把这个数据直接还给服务器，验证客户端发给服务器的情况

SendMsg(ab);

// 数据刚刚好，没有多余的

if (length + lengthLength == data_buff.Length)

{

data_buff = null;

}

else

{

// 数据有剩余，保存剩余数据，等下一个Update解析

byte[] t = new byte[data_buff.Length - length - lengthLength];

Array.Copy(data_buff, lengthLength + length, t, 0, t.Length);

data_buff = t;

}

}

}

}

// 发送数据

public void SendMsg( pb::IMessage<AddPerson> message)

{

if (outputStream != null)

{

// WriteMessage 里面会先write一个长度，然后再write真实数据

//// outputStream.WriteMessage(message);

outputStream.Flush(); // 把buffer数据写入到tcpClient的流里面

}

}

public void ReceiveMessage(IAsyncResult ar)

{

Debug.Log("消息：" + receive_buff);

try

{

// 本次接收到的数据长度

int bytesRead = tcpClient.GetStream().EndRead(ar);

if (bytesRead < 1)

{

Debug.LogError("bytesRead < 1");

return;

}

else

{

if (data_buff == null)

{

// buff里面没有数据

data_buff = new byte[bytesRead];

Array.Copy(receive_buff, data_buff, bytesRead);

}

else

{

// buff里面有数据，要和新数据整合起来

byte[] new_data = new byte[bytesRead + data_buff.Length];

Array.Copy(data_buff, new_data, data_buff.Length);

Array.Copy(receive_buff, 0, new_data, data_buff.Length, bytesRead);

data_buff = new_data;

}

}

// 继续监听下一波数据

tcpClient.GetStream().BeginRead(receive_buff, 0, tcpClient.ReceiveBufferSize, ReceiveMessage, null);

}

catch (Exception ex)

{

// 为了防止报ex没被使用的警告

Debug.Log(ex);

}

}

}

处理与Netty服务器通信的粘包、拆包

服务器的粘包拆包是Netty本身支持的解码编码器，如下图

服务器粘包、拆包处理方式

总共四行，其中第一行作用在拆包的时候，第三行作用在粘包的时候（我猜的）。

它这个拆包粘包不是普通的那种固定4个字节标示长度的，而是有时候1个字节，有时候是2、3、4、5个字节，根据当前发送的真实数据的长度定的。

在普通的方案粘包方案，数据是这样的：4个字节+真实数据

有的是用换行回车作为标识符拆包、粘包

那在Netty的方案里，包长度究竟是几个字节呢？

其实它也是用到了Protobuff里面的数据读取、保存方式，感兴趣的可以打开protobuf3-for-unity-3.0.0\src\Google.Protobuf.sln工程看一下，在Google.Protobuf项目中，打开CodedInputStream.cs：

SlowReadRawVarint32

包头占用几个字节是由下面这个函数计算的：

这个是计算一个uint数据的真实长度的方法

这也是protobuff对象编码后数据会比较小的主要原因。比如一个对象编码后得到的是440个字节数据，那么调用ComputeRawVarint32Size(440)的返回值是2，也就是服务器和客户端发送的数据最终长度是440+2=442个字节。明白了这些，拆包和粘包就都不是问题了。

上面的代码里，粘包是这一段：

public void SendMsg(IMessage message)
    {
        if (outputStream != null)
        {
            // WriteMessage 里面会先write一个长度，然后再write真实数据
            outputStream.WriteMessage(message);
            outputStream.Flush();       // 把buffer数据写入到tcpClient的流里面
        }
    }

乍一看，好像没有在真实数据前面加长度啊？其实，在outputStream的WriteMessage里面已经有WriteLength了，帮我们做好了。

image.png

再看拆包：

            // 把数据传给CodedInputStream计算本次包的长度
            CodedInputStream inputStream = new CodedInputStream(data_buff);
            int length = inputStream.ReadLength();
            // 计算"包长度"占用的字节数，后面取数据的时候扣掉这个字节数，就是真实数据长度
            int lengthLength = CodedOutputStream.ComputeLengthSize(length);

            // 当前数据足够解析一个包了
            if (length + lengthLength <= data_buff.Length)
            {
                byte[] real_data = new byte[length];
                // 拷贝真实数据
                Array.Copy(data_buff, lengthLength, real_data, 0, length);

                // 假设服务器给你发了个AddressBook
                AddressBook ab = AddressBook.Parser.ParseFrom(real_data);
                ...
            }

先用CodedInputStream 看看这个“包大小”值是多少，再用CodedOutputStream.ComputeLengthSize计算这个“包大小”占几个字节，然后就明白真实数据从哪里开始，占多少字节了。

自己写的消息注册和分发：

using Google.Protobuf;

using Google.Protobuf.WellKnownTypes;

using System;

using System.Collections;

using System.Collections.Generic;

using System.Net.Sockets;

using UnityEngine;

using ARProto;

using pb = global::Google.Protobuf;

using System.Threading;

using System.Text;

using NIO;

public class ClientManager : MonoBehaviour

{

private static object lockObj = new object();

public static ClientManager instance;

public bool isEditIP;

public string IP = "106.2.124.243";

public int PORT = 9090;

public delegate void ResponMsgHandler(ServerMessage msg);

public delegate void BroadcastHandler(ServerMessage msg);

Queue<ServerMessage> QueueServerMessage = new Queue<ServerMessage>();

public event BroadcastHandler broadcastHandler;

private Dictionary<int, ResponMsgHandler> mDicMsgs = new Dictionary<int, ResponMsgHandler>();

private List<BroadcastHandler> BroadMsg = new List<BroadcastHandler>();

/// <summary>

/// 发送请求消息

/// </summary>

/// <param name="clientMessage">消息</param>

/// <param name="handler">被监听回调方法</param>

public void SendRequestMsg(ClientMessage clientMessage, ResponMsgHandler handler)

{

AddListener(clientMessage.Seq, handler);

SendMsg(clientMessage);

}

/// <summary>

/// 订阅广播消息只需要注册一遍

/// </summary>

/// <param name="handler">被监听方法</param>

public void AddBroadcastListener(BroadcastHandler handler)

{

if (BroadMsg == null)

BroadMsg = new List<BroadcastHandler>();

if (BroadMsg.Contains(handler))

{

return;

}

else

{

BroadMsg.Add(handler);

broadcastHandler += handler;

}

/// <summary>

/// 取消对参数handler的监听

/// </summary>

/// <param name="msgType">消息类型</param>

/// <param name="handler">被监听方法</param>

public void RemoveListener(int msgType, ResponMsgHandler handler)

{

if (mDicMsgs != null && mDicMsgs.ContainsKey(msgType))

{

mDicMsgs[msgType] -= handler;

mDicMsgs.Remove(msgType);

}

private void AddListener(int msgType, ResponMsgHandler handler)

{

if (mDicMsgs == null)

mDicMsgs = new Dictionary<int, ResponMsgHandler>();

if (mDicMsgs.ContainsKey(msgType))

{

return;

}

else

{

mDicMsgs.Add(msgType, null);

mDicMsgs[msgType] += handler;

}

private void ClearAllListeners()

{

if (mDicMsgs != null)

mDicMsgs.Clear();

}

private void SendResponMsg(int msgType, ServerMessage msg)

{

ResponMsgHandler handler;

if (mDicMsgs != null && mDicMsgs.TryGetValue(msgType, out handler))

{

if (handler != null)

handler(msg);

}

private void SendBroadcastMsg(ServerMessage msg)

{

if (broadcastHandler != null)

broadcastHandler(msg);

}

void Awake()

{

instance = this;

}

Thread ThreadDoMessage;

bool isThreadDoMessageAbort;

void Start()

{

if (!isEditIP)

{

string str = Callback.UnityCheckEnvironment();

Debug.Log("UnityCheckEnvironment str:" + str);

if (str.Equals("dev") || str.Equals("test"))

{

IP = "106.2.124.243";

PORT = 29090;

}

if (str.Equals("online"))

{

IP = "106.2.124.243";

PORT = 9090;

}

StartConnect(IP, PORT);

ThreadDoMessage = new Thread(new ThreadStart(DoMessage));

//ThreadDoMessage.IsBackground = true;

//ThreadDoMessage.Priority = System.Threading.ThreadPriority.Normal;

ThreadDoMessage.Start();

isThreadDoMessageAbort = true;

InvokeRepeating("SendRepeat", 0, 2);

}

TcpClient tcpClient;

byte[] receive_buff;

byte[] data_buff;

byte[] data_buff2;

CodedOutputStream outputStream;

bool StartConnect(string IP, int PORT)

{

TcpClient client = new TcpClient();

tcpClient = client;

try

{

client.Connect(IP, PORT);

receive_buff = new byte[client.ReceiveBufferSize];

outputStream = new CodedOutputStream(client.GetStream());

client.GetStream().BeginRead(receive_buff, 0, client.ReceiveBufferSize, new AsyncCallback(ReceiveMessage), null);

}

catch (Exception e)

{

Debug.LogError("Connect Error:" + e);

}

isSuccess = client.Connected;

QueueServerMessage.Clear();

Debug.Log("StartConnect isSuccess:" + client.Connected);

heartTime = 0;

return isSuccess;

}

int heartTime = 0;

void SendRepeat()

{

SendMsg(PackMessage.PackClientHearMessage());

heartTime++;

// Debug.Log("SendMsg PackMessage.PackClientHearMessage:" + heartTime);

}

private bool SocketConnected()

{

try

{

return !tcpClient.Client.Poll(1, SelectMode.SelectRead) && (tcpClient.Client.Available == 0);

}

catch (SocketException)

{

return false;

}

catch (ObjectDisposedException)

{

return false;

}

void FixedUpdate()

{

if (!isSuccess)

{

Debug.Log("ReconnectUI");

ReconnectUI();

// return;

}

while (QueueServerMessage.Count > 0)

{

ServerMessageDo(QueueServerMessage.Dequeue());

}

void DoMessage()

{

Debug.Log("DoMessage 1");

while (true)

{

try

{

if (!isThreadDoMessageAbort)

{

Debug.Log("Exit DoMessage");

return;

}

// Debug.Log("DoMessage while (true)");

// if (((tcpClient.Client.Poll(1000, SelectMode.SelectRead) && (tcpClient.Client.Available == 0)) || !tcpClient.Client.Connected))

// {

// Debug.Log("tcpClient false");

// isSuccess = false;

// }

// isSuccess=SocketConnected();

lock (lockObj)

{

if (null != tcpClient && null != tcpClient.Client)

{

if (!tcpClient.Client.Connected || heartTime >= 2)

{

// Debug.Log("Update heartTime:" + heartTime);

isSuccess = false;

}

// if (!isSuccess)

// {

// Debug.Log("!isSuccess");

// // ReconnectUI();

// return;

// }

while (true)

{

if (data_buff != null)

{

lock (lockObj)

{

if (data_buff2 == null)

{

data_buff2 = new byte[data_buff.Length];

Array.Copy(data_buff, data_buff2, data_buff.Length);

}

else

{

byte[] temp = new byte[data_buff2.Length + data_buff.Length];

Array.Copy(data_buff2, temp, data_buff2.Length);

Array.Copy(data_buff, 0, temp, data_buff2.Length, data_buff.Length);

data_buff2 = temp;

}

data_buff = null;

}

else

{

break;

}

while (true)

{

if (data_buff2 != null)

{

CodedInputStream inputStream = new CodedInputStream(data_buff2);

int length = inputStream.ReadLength();

int lengthLength = CodedOutputStream.ComputeLengthSize(length);

if (length + lengthLength <= data_buff2.Length)

{

byte[] real_data = new byte[length];

Array.Copy(data_buff2, lengthLength, real_data, 0, length);

if (real_data.Length == 0)

{

Debug.LogError("real_data.Length==0");

}

ServerMessage sMsg = ServerMessage.Parser.ParseFrom(real_data);

// ServerMessageDo(sMsg);

QueueServerMessage.Enqueue(sMsg);

// Debug.Log("--ServerMessage:" + sMsg);

// Write.Log("--ServerMessage:" + sMsg);

if (length + lengthLength == data_buff2.Length)

{

data_buff2 = null;

}

else

{

byte[] temp = new byte[data_buff2.Length - length - lengthLength];

Array.Copy(data_buff2, lengthLength + length, temp, 0, temp.Length);

data_buff2 = temp;

}

else

{

break;

}

else

{

break;

}

catch (Exception exception)

{

Debug.Log("Socket exception: " + exception);

}

private void SendMsg(pb::IMessage message)

{

// Debug.Log("++SendMsg:" + message);

// Write.Log("++SendMsg:" + message);

if (!isSuccess)

{

Debug.Log("!isSuccess");

return;

}

try

{

if (outputStream != null)

{

outputStream.WriteMessage(message);

outputStream.Flush();

}

catch (Exception e)

{

Debug.LogError("SendMsg Exception:" + e);

isSuccess = false;

}

private void ServerMessageDo(ServerMessage sMsg)

{

if (sMsg == null)

{

Debug.LogError("ServerMessageDo sMsg==null");

return;

}

if (sMsg.MessageCategory == ServerMessageCategory.UndefinedServerMessageCategory)

{

Debug.LogError("sMsg.MessageCategory == ServerMessageCategory.UndefinedServerMessageCategory:" + sMsg);

return;

}

// Debug.Log("--sMsg:" + sMsg);

if (sMsg.MessageCategory == ServerMessageCategory.Broadcast)

{

// Debug.Log("Broadcast:" + sMsg);

// Write.Log("Broadcast:" + sMsg);

Broadcast broadcast = sMsg.Broadcast; // 广播

SendBroadcastMsg(sMsg);

}

else if (sMsg.MessageCategory == ServerMessageCategory.Response)

{

if (null == sMsg.Response)

{

Debug.LogError("ServerMessageDo null==sMsg.Response:" + sMsg);

return;

}

// 对客户端请求的响应

Response response = sMsg.Response;

// Debug.Log("response.Ack:" + response.Ack);

SendResponMsg(response.Ack, sMsg);

}

else if (sMsg.MessageCategory == ServerMessageCategory.ServerHeartbeat)

{

ServerHeartbeat serverHeartbeat = sMsg.ServerHeartbeat; // 服务端心跳

heartTime--;

// Debug.Log("PackClientHearMessage serverHeartbeat 心跳:" + heartTime);

}

private void ReceiveMessage(IAsyncResult ar)

{

// Debug.Log("receive_buff:" + receive_buff.Length);

try

{

// Debug.Log("ReceiveMessage id =" + Thread.CurrentThread.ManagedThreadId);

int bytesRead;

bytesRead = tcpClient.GetStream().EndRead(ar);

if (bytesRead < 1)

{

Debug.LogError("网络已断开 bytesRead < 1");

//TODO 重连

isSuccess = false;

return;

}

else

{

lock (lockObj)

{

if (data_buff == null)

{

data_buff = new byte[bytesRead];

Array.Copy(receive_buff, data_buff, bytesRead);

}

else

{

byte[] new_data = new byte[bytesRead + data_buff.Length];

Array.Copy(data_buff, new_data, data_buff.Length);

Array.Copy(receive_buff, 0, new_data, data_buff.Length, bytesRead);

data_buff = new_data;

}

// Debug.Log("ThreadId:" + Thread.CurrentThread.ManagedThreadId + "data_buff:" + data_buff.Length);

}

tcpClient.GetStream().BeginRead(receive_buff, 0, tcpClient.ReceiveBufferSize, new AsyncCallback(ReceiveMessage), null);

//string strbuff = Encoding.UTF8.GetString(receive_buff, 0, bytesRead);

// Debug.Log("bytesRead:" + bytesRead + "/" + receive_buff.Length + " receive_buff:" + strbuff);

// Write.Log("bytesRead:" + bytesRead + "/" + receive_buff.Length + " receive_buff:" + strbuff);

}

catch (Exception ex)

{

Debug.Log("ReceiveMessageException:" + ex);

}

// int reconnetTime = 0;

bool isSuccess = false;

public void ReconnectUI()

{

ARCarUIManager.Instance.ARNetPanel.SetActive(true);

}

public bool Reconnect()

{

Debug.Log("正在尝试重新连接网络...");

bool isok = StartConnect(IP, PORT);

ARCarUIManager.Instance.ARNetPanel.SetActive(false);

return isok;

}

// void Reconnect()

// {

// Debug.Log("Reconnect");

// if (!isSuccess)

// {

// tcpClient.Close();

// InvokeRepeating("InvokeReconnect", 0, 8);

// }

// void InvokeReconnect()

// {

// reconnetTime++;

// Debug.LogError("正在尝试重新连接网络..." + reconnetTime);

// StartConnect(IP, PORT);

// if (isSuccess || reconnetTime >= 5)

// {

// CancelInvoke("InvokeReconnect");

// }

public void OnDestroy()

{

try

{

tcpClient.GetStream().Close();

tcpClient.Close();

isSuccess = false;

ThreadDoMessage.Abort();

isThreadDoMessageAbort = false;

}

catch (Exception e)

{

Debug.Log(" tcpClient.Close e:" + e);

}

broadcastHandler = null;

mDicMsgs = null;

BroadMsg = null;

ClientManager.instance = null;

}

巴特西

ProtoBuff3 unity_TCP网络发包解包&&消息订阅

处理与Netty服务器通信的粘包、拆包

最新文章

热门文章