在线时间:8:00-16:00
迪恩网络APP
随时随地掌握行业动态
扫描二维码
关注迪恩网络微信公众号
假设消息头为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 } |
2023-10-27
2022-08-15
2022-08-17
2022-09-23
2022-08-13
请发表评论