推荐订阅源

假设消息头为16byte

0-3   消息头标识

4-7   消息体长度

8-15 CRC校验码

public class StreamHelper
    {
        #region 将消息转为包含消息头的字节数组
        /// <summary>
        /// 将消息转为包含消息头的字节数组
        /// </summary>
        /// <param name="stream"></param>
        /// <param name="message"></param>
        /// <returns></returns>
        public static byte[] GetWrappedMessage(Google.ProtocolBuffers.IMessage message)
        {
            byte[] buff = message.ToByteArray();
            CRC32 crc32 = new CRC32();
            crc32.Update(buff);

            MemoryStream ms = new MemoryStream();
            byte[] tag = BitConverter.GetBytes(Constants.Tag);
            byte[] len = BitConverter.GetBytes(buff.Length);
            byte[] crc = BitConverter.GetBytes(crc32.Value);

            ms.Write(tag, 0, tag.Length); // 写入标头
            ms.Write(len, 0, len.Length); // 写入消息长度
            ms.Write(crc, 0, crc.Length); // 写入crc32值
            ms.Write(buff, 0, buff.Length);

            byte[] result = ms.ToArray();
            ms.Close();

            return result;
        }
        #endregion

        #region 将消息写入流
        /// <summary>
        /// 将消息写入流
        /// </summary>
        /// <param name="stream"></param>
        /// <returns></returns>
        public static void WriteToStream(NetworkStream stream, Google.ProtocolBuffers.IMessage message)
        {
            byte[] buff = message.ToByteArray();
            CRC32 crc32 = new CRC32();
            crc32.Update(buff);

            byte[] tag = BitConverter.GetBytes(Constants.Tag);
            byte[] len = BitConverter.GetBytes(buff.Length);
            byte[] crc = BitConverter.GetBytes(crc32.Value);

            stream.Write(tag, 0, tag.Length); // 写入标头
            stream.Write(len, 0, len.Length); // 写入消息长度
            stream.Write(crc, 0, crc.Length); // 写入crc32值
            stream.Write(buff, 0, buff.Length);
        }
        #endregion

        #region 将消息写入Socket
        /// <summary>
        /// 将消息写入Socket
        /// </summary>
        /// <param name="socket"></param>
        /// <returns></returns>
        public static void WriteToSocket(Socket socket, Google.ProtocolBuffers.IMessage message)
        {
            byte[] buff = message.ToByteArray();
            CRC32 crc32 = new CRC32();
            crc32.Update(buff);

            byte[] tag = BitConverter.GetBytes(Constants.Tag);
            byte[] len = BitConverter.GetBytes(buff.Length);
            byte[] crc = BitConverter.GetBytes(crc32.Value);

            socket.SendBufferSize = Constants.HeadSize + buff.Length;

            socket.Send(tag); // 写入标头
            socket.Send(len); // 写入消息长度
            socket.Send(crc); // 写入crc32值
            socket.Send(buff);
        }
        #endregion

        #region 从NetworkStream读出所有字节
        /// <summary>
        /// 从NetworkStream读出所有字节
        /// </summary>
        /// <param name="stream"></param>
        /// <returns></returns>
        public static byte[] ReadAllBytes(NetworkStream stream)
        {
            byte[] head = new byte[Constants.HeadSize];
            if (stream.Read(head, 0, Constants.HeadSize) != Constants.HeadSize) // 传输头16位,仅校验数据使用
            {
                return null;
            }

            HeadInfo headInfo = new HeadInfo(head);
            if (headInfo.Tag != Constants.Tag) // 标头不正确
            {
                return null;
            }

            byte[] buff = new byte[headInfo.Len];
            int size = Math.Min(buff.Length, 4096); // 一次最多读取多少个字节
            int offset = 0;
            do
            {
                size = Math.Min(size, headInfo.Len - offset);
                offset += stream.Read(buff, offset, size);
                if (offset == headInfo.Len) // 接收到指定的长度
                {
                    CRC32 crc32 = new CRC32();
                    crc32.Update(buff);
                    if (crc32.Value == headInfo.Crc32Value) // 验证CRC32
                    {
                        return buff;
                    }
                }
            }
            while (stream.DataAvailable);

            Log.ErrorFormat("客户端请求失败: offset {0}, head len {1}", offset, headInfo.Len);

            return null;
        }
        #endregion

        #region 从Socket读出所有字节
        /// <summary>
        /// 从Socket读出所有字节
        /// </summary>
        /// <param name="socket"></param>
        /// <returns></returns>
        public static byte[] ReadAllBytes(Socket socket)
        {
            byte[] head = new byte[Constants.HeadSize];
            byte[] result = null;

            SocketError error = SocketError.Success;
            int maxRepeat = 5;
            int got = 0;
            int repeat = 0;

            do // 读取标头
            {
                got += socket.Receive(head, got, Constants.HeadSize - got, SocketFlags.None, out error);
                if (got == Constants.HeadSize || repeat > maxRepeat) // 接收完全或者重复10次仍然没接收
                {
                    break;
                }
                else if (got == 0) // 获取不到数据重复的次数
                {
                    ++repeat;
                }
            }
            while (error == SocketError.Success);

            if (got != Constants.HeadSize) // 验证标头长度
            {
                Log.Error("Head长度不正确");
                return null;
            }

            HeadInfo headInfo = new HeadInfo(head);
            if (headInfo.Tag != Constants.Tag) // 标头不正确
            {
                Log.ErrorFormat("Head.Tag不正确(应为{0}, 实际{1})", Constants.Tag, headInfo.Tag);
                return null;
            }

            result = new byte[headInfo.Len];
            repeat = 0;
            got = 0;
            do // 读取消息体
            {
                got += socket.Receive(result, got, result.Length - got, SocketFlags.None, out error);
                if (got == result.Length || repeat > maxRepeat) // 接收完全或者重复10次仍然没接收
                {
                    break;
                }
                else if (got == 0) // 获取不到数据重复的次数
                {
                    ++repeat;
                }
            }
            while (error == SocketError.Success);

            if (got != result.Length) // 验证长度
            {
                Log.ErrorFormat("长度不正确(应为{0}, 实际{1})", result.Length, got);
                return null;
            }

            CRC32 crc32 = new CRC32();
            crc32.Update(result);
            if (crc32.Value != headInfo.Crc32Value) // 验证CRC32
            {
                Log.ErrorFormat("CRC校验失败(应为{0}, 实际{1})", headInfo.Crc32Value, crc32.Value);
                return null;
            }

            return result;
        }
        #endregion

        #region 异步接收Socket数据(测试用)
        /// <summary>
        /// 异步接收Socket数据
        /// </summary>
        /// <param name="socket"></param>
        /// <returns></returns>
        public static byte[] ReceiveData(Socket socket)
        {
            ReceiveObject state = new ReceiveObject() { Client = socket };
            socket.BeginReceive(state.Buffer, 0, state.Buffer.Length, SocketFlags.None, new AsyncCallback(ReadCallback), state);
            state.ReceiveDone.WaitOne(socket.ReceiveTimeout); // 超时设置

            byte[] head = new byte[Constants.HeadSize];
            state.Stream.Seek(0, SeekOrigin.Begin);
            if (state.Stream.Read(head, 0, Constants.HeadSize) != Constants.HeadSize) // 读取标头
            {
                return null;
            }

            HeadInfo headInfo = new HeadInfo(head);
            if (headInfo.Tag != Constants.Tag) // 标头不正确
            {
                return null;
            }
            if (state.Stream.Length - 16 != headInfo.Len) // 验证长度
            {
                return null;
            }

            // state.Stream.Seek(16, SeekOrigin.Begin);
            byte[] result = new byte[state.Stream.Length - 16];
            state.Stream.Read(result, 0, result.Length); // 读取消息体

            CRC32 crc32 = new CRC32();
            crc32.Update(result);
            if (crc32.Value == headInfo.Crc32Value) // 验证CRC32
            {
                return result;
            }

            return null;
        }

        /// <summary>
        /// 同步接收对象
        /// </summary>
        private class ReceiveObject : IDisposable
        {
            public Socket Client;
            public byte[] Buffer = new byte[4096];
            public System.IO.MemoryStream Stream = new System.IO.MemoryStream();
            public System.Threading.ManualResetEvent ReceiveDone = new System.Threading.ManualResetEvent(false);

            public void Dispose()
            {
                this.Stream.Close();
            }
        }

        /// <summary>
        /// read回调
        /// </summary>
        /// <param name="ar"></param>
        private static void ReadCallback(IAsyncResult ar)
        {
            ReceiveObject state = (ReceiveObject)ar.AsyncState;
            int bytesRead = state.Client.EndReceive(ar);
            if (bytesRead > 0)
            {
                try
                {
                    state.Stream.Write(state.Buffer, 0, bytesRead);
                    state.Client.BeginReceive(state.Buffer, 0, state.Buffer.Length, SocketFlags.None, new AsyncCallback(ReadCallback), state);
                }
                catch (Exception ex)
                {
                    Log.Error(ex.Message);
                    state.ReceiveDone.Set();
                }
            }
            else
            {
                state.ReceiveDone.Set();
                // state.Client.EndReceive(ar);
            }
        }
        #endregion
    }