《Unity网络游戏实战》Chapter4: 收发数据流（2）

1、介绍

上一节中解决了粘包拆包的问题，但是还有大小端兼容、发送不完整等问题需要解决，这一节就来解决这几个问题。

2、客户端

2.1 大小端兼容问题

在不同的机器上的编码方式不同，有的是大端编码，有的是小端编码。而网络数据报统一是小端编码，C#中的Reverse函数可以进行大小端编码的转换。

        // 组装消息
        byte[] bodyBytes = System.Text.Encoding.Default.GetBytes(sendStr);
        short length = (short)bodyBytes.Length;
        byte[] lenBytes = BitConverter.GetBytes(length);

        // 统一为小端编码
        if(!BitConverter.IsLittleEndian)
        {
            lenBytes.Reverse();
        }

        byte[] sendBytes = lenBytes.Concat(bodyBytes).ToArray();

2.2 发送不完整问题

由于操作系统缓冲区的设置和网络状况等问题。会出现发送不完整的情况，一个包只发了一半。

解决这个问题的方法是在发送数据前记录下要发送的数据，根据socket.Send()函数返回的发送字节数判断是否发送完整，如果没有发送完整，则继续发送。

    // 将发送数据写入发送队列
    public void Send()
    {
        // 发送数据
        string sendStr= "hello Python, hello Python, hello Python, hello Python, hello Python, hello Python";

        // 组装消息
        byte[] bodyBytes = System.Text.Encoding.Default.GetBytes(sendStr);
        short length = (short)bodyBytes.Length;
        byte[] lenBytes = BitConverter.GetBytes(length);

        // 统一为小端编码
        if(!BitConverter.IsLittleEndian)
        {
            lenBytes.Reverse();
        }

        byte[] sendBytes = lenBytes.Concat(bodyBytes).ToArray();

        // 发送钱保存数据
        ByteArray ba = new ByteArray(sendBytes);
        // 加锁，防止线程冲突
        lock(writeQueue)
        {
            writeQueue.Enqueue(ba);
        }

        // 发送数据
        DoSend();
    }

    // 读取发送队列的第一条数据，发送数据
    public void DoSend()
    {
        lock(writeQueue)
        {
            if (writeQueue.Count <= 0)
                return;
        }

        // 读取队列的第一条数据
        ByteArray ba;
        lock (writeQueue)
        {
            ba = writeQueue.First();
        }

        socket.BeginSend(ba.buffer, ba.readIndex, ba.Length, SocketFlags.None, SendCallBack, socket);
    }

    public void SendCallBack(IAsyncResult ar)
    {
        Socket socket = (Socket)ar.AsyncState;

        // 成功发送的字节
        int count = socket.EndSend(ar);

        // 取出写入队列的第一个数据
        ByteArray ba = writeQueue.First();

        ba.readIndex += count;
        if(ba.Length == 0)
        {
            lock(writeQueue)
            {
                writeQueue.Dequeue();
                ba = writeQueue.First();
            }
        }
        
        if (ba != null)
        {
            socket.BeginSend(ba.buffer, ba.readIndex, ba.Length, SocketFlags.None, SendCallBack, socket);
        }
    }

因为要记录下当前需要发送的数据的发送位置，因此实现了一个类ByteArray，记录需要发送的数据，数据长度，数据尾指针writeindex和当前发送的指针readindex。采用发送队列的形式，每次发送都是发送一条完整的数据，避免造成混乱。

2.3 缓冲区扩容

如果缓冲区剩余的空间不足（或剩下的空间很少），便需要对缓冲区进行扩容。扩容的操作和C++中Vector的扩容操作类似，申请新的空间，将为处理的数据拷贝过去。

            if(recvBuff.Remain < 16)
            {
                // 扩容之前为何要先移动缓冲区，感觉没有必要，因为在resize的时候已经是在新buffer中从0开始了
                recvBuff.MoveBytes();
                recvBuff.ReSize(recvBuff.Length * 2);
            }

原书中在扩容之前先将数据移动到开头。但是从其源码的功能来看感觉并不需要这部操作。

下面是ByteArray的代码：

using System;

public class ByteArray
{
    const int default_size = 1024;
    // 初始缓冲区大小
    int initSize = 0;
    // 缓冲区
    public byte[] buffer;
    // 已发送的索引
    public int readIndex = 0;
    // 整个数据的长度
    public int writeIndex = 0;
    // 缓冲区容量
    public int capacity = 0;
    // 缓冲区剩余空间
    public int Remain { get { return capacity - writeIndex; } }
    // 缓冲区中有效数据长度
    public int Length { get { return writeIndex - readIndex; } }

    public ByteArray(int size = default_size)
    {
        buffer = new byte[size];

        capacity = size;

        initSize = size;

        readIndex = 0;

        writeIndex = 0;
    }


    public ByteArray(byte[] defaultBytes)
    {
        buffer = defaultBytes;

        readIndex = 0;

        writeIndex = defaultBytes.Length;

        capacity = defaultBytes.Length;

        initSize = defaultBytes.Length;
    }

    // 扩容
    public void ReSize(int size)
    {
        if (size < Length || size < initSize)
            return;
        // 指数扩容：从2的倍数中找一个比原来大的
        int n = 1;
        while (n < size)
            n *= 2;
        // 申请新数组并拷贝数据
        capacity = n;
        byte[] newbuffer = new byte[capacity];
        Array.Copy(buffer, readIndex, newbuffer, 0, Length);
        // buffer指向newbuffer
        buffer = newbuffer;
        writeIndex = Length;
        readIndex = 0;
    }

    // 向缓冲区写数据, count是要写入缓冲区的字节数
    public int Write(byte[] bs, int offset, int count)
    {
        if (Remain < count)
            ReSize(Length + count);

        Array.Copy(bs, offset, buffer, writeIndex, count);
        writeIndex += count;
        return count;
    }

    // 读取缓冲区中的数据 将缓冲区中的count字节数据读取到bs中
    public int Read(byte[] bs, int offset, int count)
    {
        count = Math.Min(count, Length);
        Array.Copy(buffer, readIndex, bs, offset, count);
        readIndex += count;
        CheckAndMoveBytes();
        return count;
    }

    //检查并移动数据
    public void CheckAndMoveBytes()
    {
        if (Length < 128)
        {
            MoveBytes();
        }
    }

    //移动数据
    public void MoveBytes()
    {
        Array.Copy(buffer, readIndex, buffer, 0, Length);
        writeIndex = Length;
        readIndex = 0;
    }

    public short ReadInt16()
    {
        if (Length < 2)
            return 0;

        short ret = BitConverter.ToInt16(buffer, readIndex);
        readIndex += 2;
        CheckAndMoveBytes();
        return ret;
    }

    public int ReadInt32()
    {
        if (Length < 4)
            return 0;

        int ret = BitConverter.ToInt32(buffer, readIndex);
        readIndex += 4;
        CheckAndMoveBytes();
        return ret;
    }

    //打印缓冲区
    public override string ToString()
    {
        return BitConverter.ToString(buffer, readIndex, Length);
    }

    //打印调试信息
    public string Debug()
    {
        return string.Format("readIdx({0}) writeIdx({1}) bytes({2})",
            readIndex,
            writeIndex,
            BitConverter.ToString(buffer, 0, capacity)
        );
    }
}

客户端完整代码：

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

public class Echo : MonoBehaviour
{
    Socket socket;

    ByteArray recvBuff = new ByteArray();

    int headerSize = 4;

    bool startSendMsg = false;

    // 发送队列
    Queue<ByteArray> writeQueue = new Queue<ByteArray>();

    private void Start()
    {
        socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        socket.Connect("127.0.0.1", 8888);
        socket.BeginReceive(recvBuff.buffer, recvBuff.writeIndex, 
            recvBuff.Remain, SocketFlags.None, ReceiveCallBack, socket);
    }

    void ReceiveCallBack(IAsyncResult ar)
    {
        try
        {
            Socket socket = (Socket)ar.AsyncState;
            // 接收到的数据长度
            int count = socket.EndReceive(ar);

            recvBuff.writeIndex += count;
            // 处理消息
            HandleReceiveData();
            // 继续接收数据
            if(recvBuff.Remain < 16)
            {
                // 扩容之前为何要先移动缓冲区，感觉没有必要，因为在resize的时候已经是在新buffer中从0开始了
                recvBuff.MoveBytes();
                recvBuff.ReSize(recvBuff.Length * 2);
            }
            socket.BeginReceive(recvBuff.buffer, recvBuff.writeIndex, 
                recvBuff.Remain, SocketFlags.None, ReceiveCallBack, socket);
        }
        catch (SocketException ex)
        {
            Debug.Log("socket receive failed" + ex.ToString());
        }
    }

    void HandleReceiveData()
    {
        Debug.Log("接收缓冲区字节数 =" + recvBuff.Length);

        // 检查数据长度是否超过包头
        if (recvBuff.Length < headerSize)
            return;

        // 读取包头长度 -- 从缓冲区的开头读取4字节, 两种写法都可以
        int bodyLength = BitConverter.ToInt32(recvBuff.buffer, recvBuff.readIndex);
        
        //int readInx = recvBuff.readIndex;
        //byte[] bytes = recvBuff.buffer;
        //int bodyLength = (bytes[readInx + 3] << 24) | bytes[readInx + 2] << 16 | 
        //                        bytes[readInx + 1] << 8 | bytes[readInx];

        // 检查是否足够一条信息
        if (recvBuff.Length < headerSize + bodyLength)
            return;

        // 移动readindex一个包头的字节数
        recvBuff.readIndex += headerSize;

        // 拷贝一条数据到临时byte数组中
        byte[] tmpBytes = new byte[bodyLength];
        recvBuff.Read(tmpBytes, 0, bodyLength);

        // 处理一条消息
        string data = System.Text.Encoding.UTF8.GetString(tmpBytes);
        Debug.Log("receive data = " + data);
        Debug.Log("capacity = " + recvBuff.capacity);

        // 继续处理消息
        HandleReceiveData();
    }

    // 将发送数据写入发送队列
    public void Send()
    {
        // 发送数据
        string sendStr= "hello Python, hello Python, hello Python, hello Python, hello Python, hello Python";

        // 组装消息
        byte[] bodyBytes = System.Text.Encoding.Default.GetBytes(sendStr);
        short length = (short)bodyBytes.Length;
        byte[] lenBytes = BitConverter.GetBytes(length);

        // 统一为小端编码
        if(!BitConverter.IsLittleEndian)
        {
            lenBytes.Reverse();
        }

        byte[] sendBytes = lenBytes.Concat(bodyBytes).ToArray();

        // 发送钱保存数据
        ByteArray ba = new ByteArray(sendBytes);
        // 加锁，防止线程冲突
        lock(writeQueue)
        {
            writeQueue.Enqueue(ba);
        }

        // 发送数据
        DoSend();
    }

    // 读取发送队列的第一条数据，发送数据
    public void DoSend()
    {
        lock(writeQueue)
        {
            if (writeQueue.Count <= 0)
                return;
        }

        // 读取队列的第一条数据
        ByteArray ba;
        lock (writeQueue)
        {
            ba = writeQueue.First();
        }

        socket.BeginSend(ba.buffer, ba.readIndex, ba.Length, SocketFlags.None, SendCallBack, socket);
    }

    public void SendCallBack(IAsyncResult ar)
    {
        Socket socket = (Socket)ar.AsyncState;

        // 成功发送的字节
        int count = socket.EndSend(ar);

        // 取出写入队列的第一个数据
        ByteArray ba = writeQueue.First();

        ba.readIndex += count;
        if(ba.Length == 0)
        {
            lock(writeQueue)
            {
                writeQueue.Dequeue();
                ba = writeQueue.First();
            }
        }
        
        if (ba != null)
        {
            socket.BeginSend(ba.buffer, ba.readIndex, ba.Length, SocketFlags.None, SendCallBack, socket);
        }
    }

    public void OnButtonStartClick()
    {
        startSendMsg = true;
    }

    private void Update()
    {
        if (!startSendMsg)
            return;

        Send();
    }

}

3、服务端

服务端的代码与上一节没什么差别。

# Handle The TCP problem
import socket
import time
import struct

def parse_msg_length(head):
    # get message body length
    length = 0
    for i in range(len(head)):
        length += ord(head[i]) * 256 ** i
    return length

def handle_message(read_buffer):
    # check buffer > head size
    buffer_count = len(read_buffer)
    if buffer_count < head_size:
        return False, 0
    # check buffer > one whole data
    msg_body_size = parse_msg_length(read_buffer[0:head_size])
    if buffer_count < head_size + msg_body_size:
        return False, 0
    # handle message
    data = read_buffer[head_size:msg_body_size+head_size]
    print 'data:', data
    return True, msg_body_size

# server socket
ip = "127.0.0.1"
port = 8888
server_address = (ip, port)
buffer_size = 1024
head_size = 2
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(server_address)
server.listen(1)
read_buffer = ""

print "Waiting for Connection."
while True:
    connection, client_address = server.accept()
    print "client connected: ", client_address
    while True:
        try:
            msg_body = "Hello Unity Hello Unity Hello Unity Hello Unity"
            pack_format = 'I' + str(len(msg_body)) + 's'
            send_data = (len(msg_body), msg_body)
            send_bytes = struct.pack(pack_format, *send_data)
            connection.send(send_bytes)
            # time.sleep(1)
            receive_data = connection.recv(buffer_size)
            if receive_data:
                read_buffer += receive_data
                result, length = handle_message(read_buffer)
                if result:
                    read_buffer = read_buffer[head_size + length:]
                else:
                    pass
                print 'read_buffer: ', read_buffer
            else:
                connection.close()
                break
        except socket.error:
            connection.close()

4、效果

将服务端代码放在云运行，本地运行客户端：


image.png

image.png

在服务端中缓冲区read_buffer中的数据很多，但是可以正常解析出数据包。服务器的接收和发送都没有异常。

将服务端的接收消息注释掉，服务端不停给客户端发送数据：


image.png

image.png

可以看到，客户端的接收处理数据都正常。当缓冲区数据有大于一条数据便一直处理，直到剩余数据不足一条。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,684评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 87,143评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 151,214评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,788评论 1赞 277
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,796评论 5赞 368
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,665评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,027评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,679评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 41,346评论 1赞 299
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,664评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,766评论 1赞 331
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,412评论 4赞 321
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,015评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,974评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,203评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,073评论 2赞 350
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,501评论 2赞 343