您所在的位置：程序员俱乐部 > 编程开发 > .NET > 项目笔记---C#异步Socket示例

项目笔记---C#异步Socket示例

2014/11/27 16:23:33 史蒂芬King 程序员俱乐部我要评论(0)

摘要：概要在C#领域或者说.net通信领域中有着众多的解决方案，WCF，HttpRequest，WebAPI，Remoting，socket等技术。这些技术都有着自己擅长的领域，或者被合并或者仍然应用于某些场合。本文主要介绍Socket通讯，因其有着跨平台、跨语言、高性能等优势，适合某些情况的应用以及性能优越的解决方案。本文是基于一个小项目中的应用，使用了异步方式的Socket通讯，性能上达到多客户端多点通讯，大文件（M-G级别）的文件传输，异步长连接上的性能优势，但此项目也有些不足
标签：笔记 C# 异步socket 项目 socket 异步

概要

在C#领域或者说.net通信领域中有着众多的解决方案，WCF，HttpRequest，WebAPI，Remoting，socket等技术。这些技术都有着自己擅长的领域，或者被合并或者仍然应用于某些场合。本文主要介绍Socket通讯，因其有着跨平台、跨语言、高性能等优势，适合某些情况的应用以及性能优越的解决方案。

本文是基于一个小项目中的应用，使用了异步方式的Socket通讯，性能上达到多客户端多点通讯，大文件（M-G级别）的文件传输，异步长连接上的性能优势，但此项目也有些不足：未进行大量的外网长时间传输测试，对不稳定的网络状况未做很好的处理，例如加入断点续传功能，对Socket传输错误（其中微软提供了大量的Socket通讯错误代码指示错误类型，请参考http://www.cnblogs.com/Aricc/archive/2010/01/29/1659134.html）未做一个很好的分类处理，只简单的统一为一个类型的错误。所以，此项目介绍只是想抛砖引玉介绍异步Socket通讯，如果有不足或改进之处，还请各位不吝指出。

同步和异步区别

这里的同步和异步指的是服务端Accept接受客户端连接请求的方式。在同步模式下，服务端要开启一个Thread线程循环监听可能来自客户端的服务，如果没有则阻塞，如果有连接则接受连接并存入Connection Pool连接池，这样一个连接（或者说一个连接线程，一般占用系统内存约为2M，具体原因请参考《CLR via C#》中线程章节），这样32位系统下单一应用程序最大内存不超过2G，也就是说，单一连接服务端所能接受客户端最大的请求数为1000（实测下为700+）；而异步连接则应用非阻塞式异步连接机制（http://www.cnblogs.com/2018/archive/2011/05/10/2040333.html）BeginAccept异步接受客户端请求并执行相应请求逻辑，在执行完毕后系统能自动优化，并当再次应用时唤醒，从而达到可接受大量的客户端请求，但也限于“同时”执行的客户端数量，对于某些长连接客户端巨大，但并发性小的情景适用。

自定义 协议

众所周知，在Socket通讯中传输的普通的字符串或者二进制数据流，不适用于一些复杂的情况，例如约定一个可扩展的协议，可变长协议等，所以本项目采用自定义协议类型来满足可扩展需求：

Header

协议头

命令

流1长度

流2长度

2字节

4字节

Body

流1

流2

N字节

说明：

协议头为：FF 7E ，2字节固定值

命令为：命令编号，如1001

流1长度：Int32型指示Body中的流1的长度

流2长度：Int32型指示Body中的流2的长度

流1：字节流

流2：字节流

这样，基于上述协议可自定义流1、流2的长度分别存放不同数据，基于协议还可以对数据协议进行封装，提供公共的解析方式。

class="code_img_closed" src="/Upload/Images/2014112716/0015B68B3C38AA5B.gif" alt="" />

  1     /// <summary>
  2     /// 通讯二进制协议，此协议基于变长的流传输。
  3     /// 注：扩展此方法成员时，请重写相关方法。
  4     /// </summary>
  5     /// <remarks>
  6     /// Create By CYS
  7     /// </remarks>
  8     public class CommunicateProtocol : IDisposable
  9     {
 10         #region Public Properties
 11         /// <summary>
 12         /// Byte array length of flag
 13         /// </summary>
 14         public const int ByteLength_HeaderFlag = 2;
 15         /// <summary>
 16         /// Byte array length of command
 17         /// </summary>
 18         public const int ByteLength_HeaderCmd = 4;
 19         /// <summary>
 20         /// Byte array length of header stream1
 21         /// </summary>
 22         public const int ByteLength_HeaderStream1Len = 4;
 23         /// <summary>
 24         /// Byte array length of header stream2
 25         /// </summary>
 26         public const int ByteLength_HeaderStream2Len = 4;
 27         /// <summary>
 28         /// 协议头长度
 29         /// </summary>
 30         public static int FlagLen
 31         {
 32             get { return ByteLength_HeaderFlag; }
 33         }
 34         /// <summary>
 35         /// 命令（Int32）
 36         /// </summary>
 37         public int Command
 38         {
 39             get
 40             {
 41                 return BitConverter.ToInt32(header_Cmd, 0);
 42             }
 43             set
 44             {
 45                 BitConverter.GetBytes(value).CopyTo(header_Cmd, 0);
 46             }
 47         }
 48         /// <summary>
 49         /// 流1长度
 50         /// </summary>
 51         /// <returns></returns>
 52         public int Stream1Len
 53         {
 54             get
 55             {
 56                 return BitConverter.ToInt32(header_Stream1Len, 0);
 57             }
 58             set
 59             {
 60                 BitConverter.GetBytes(value).CopyTo(header_Stream1Len, 0);
 61             }
 62         }
 63         /// <summary>
 64         /// 流2长度
 65         /// </summary>
 66         /// <returns></returns>
 67         public int Stream2Len
 68         {
 69             get
 70             {
 71                 return BitConverter.ToInt32(header_Stream2Len, 0);
 72             }
 73             set
 74             {
 75                 BitConverter.GetBytes(value).CopyTo(header_Stream2Len, 0);
 76             }
 77         }
 78         #endregion Public Properties
 79 
 80         #region Private Properties
 81         private static byte[] header_Flag = new byte[ByteLength_HeaderFlag];
 82         private byte[] header_Cmd = new byte[ByteLength_HeaderCmd];
 83         private byte[] header_Stream1Len = new byte[ByteLength_HeaderStream1Len];
 84         private byte[] header_Stream2Len = new byte[ByteLength_HeaderStream1Len];
 85         private byte[] body_Stream1 = new byte[0];
 86         private Stream body_Stream2;
 87         #endregion Private Properties
 88 
 89         #region Constructor
 90         /// <summary>
 91         /// Static constructor
 92         /// </summary>
 93         static CommunicateProtocol()
 94         {
 95             header_Flag = new byte[ByteLength_HeaderFlag] { 0xFF, 0x7E };
 96         }
 97 
 98         #endregion Constructor
 99 
100         #region Public Method
101         /// <summary>
102         /// 判断是否是协议头标志
103         /// </summary>
104         /// <param name="bytes"></param>
105         /// <returns></returns>
106         public static bool CheckFlag(byte[] bytes)
107         {
108             if (bytes.Length != header_Flag.Length)
109                 return false;
110             if (bytes.Length != 2)
111                 return false;
112             if (!bytes[0].Equals(header_Flag[0]) || !bytes[1].Equals(header_Flag[1]))
113                 return false;
114             return true;
115         }
116         /// <summary>
117         /// SetStream1
118         /// </summary>
119         /// <param name="sm"></param>
120         public void SetStream1(byte[] sm)
121         {
122             body_Stream1 = sm;
123         }
124         /// <summary>
125         /// GetStream1
126         /// </summary>
127         /// <returns></returns>
128         public byte[] GetStream1()
129         {
130             return body_Stream1;
131         }
132         /// <summary>
133         /// SetStream2
134         /// </summary>
135         /// <param name="sm"></param>
136         public void SetStream2(Stream sm)
137         {
138             body_Stream2 = sm;
139         }
140         /// <summary>
141         /// body_Stream2
142         /// </summary>
143         /// <returns></returns>
144         public Stream GetStream2()
145         {
146             return body_Stream2;
147         }
148         /// <summary>
149         /// GetHeaderBytes
150         /// </summary>
151         /// <returns></returns>
152         public byte[] GetHeaderBytes()
153         {
154             int offset = 0;
155             byte[] bytes = new byte[ByteLength_HeaderFlag + ByteLength_HeaderCmd + ByteLength_HeaderStream1Len + ByteLength_HeaderStream2Len];
156 
157             Array.Copy(header_Flag, 0, bytes, 0, ByteLength_HeaderFlag); offset += ByteLength_HeaderFlag;
158             Array.Copy(header_Cmd, 0, bytes, offset, ByteLength_HeaderCmd); offset += ByteLength_HeaderCmd;
159             Array.Copy(header_Stream1Len, 0, bytes, offset, ByteLength_HeaderStream1Len); offset += ByteLength_HeaderStream1Len;
160             Array.Copy(header_Stream2Len, 0, bytes, offset, ByteLength_HeaderStream2Len); offset += ByteLength_HeaderStream2Len;
161 
162             return bytes;
163         }
164         /// <summary>
165         /// InitProtocolHeader
166         /// </summary>
167         /// <returns></returns>
168         public static CommunicateProtocol InitProtocolHeader(byte[] source)
169         {
170             if (source.Length < ByteLength_HeaderCmd + ByteLength_HeaderStream1Len + ByteLength_HeaderStream2Len)
171             {
172                 throw new Exception("byte length is illegal");
173             }
174 
175             byte[] header_cmd = new byte[ByteLength_HeaderCmd];
176             byte[] header_stream1len = new byte[ByteLength_HeaderStream1Len];
177             byte[] header_stream2len = new byte[ByteLength_HeaderStream2Len];
178             Array.Copy(source, 0, header_cmd, 0, ByteLength_HeaderCmd);
179             Array.Copy(source, ByteLength_HeaderCmd, header_stream1len, 0, ByteLength_HeaderStream1Len);
180             Array.Copy(source, ByteLength_HeaderCmd + ByteLength_HeaderStream1Len, header_stream2len, 0, ByteLength_HeaderStream2Len);
181 
182             return new CommunicateProtocol
183             {
184                 Command = BitConverter.ToInt32(header_cmd, 0),
185                 Stream1Len = BitConverter.ToInt32(header_stream1len, 0),
186                 Stream2Len = BitConverter.ToInt32(header_stream2len, 0),
187             };
188         }
189         #endregion Public Method
190 
191         #region Private Method
192 
193         #endregion Private Method
194 
195         #region IDisposable 成员
196         /// <summary>
197         /// Dispose
198         /// </summary>
199         public void Dispose()
200         {
201             header_Cmd = null;
202             header_Stream1Len = null;
203             header_Stream2Len = null;
204             body_Stream1 = null;
205             body_Stream2 = null;
206         }
207 
208         #endregion
209     }

View Code

通讯机制

传统意义上的上传与下载请求就是，一端发起Request请求，另一端接受并答复请求内容，这样就完成了一次请求应答。然而，如果要实现更多的控制功能，就要在这“一去一回”上增加通信应答次数，类似于TCP的三次握手请求。

其中，当请求被拒绝时，应向请求端发送错误答复998，这样就可以在这个过程中建立一个完善的请求答复机制。

 1     public abstract class ContractAdapter
 2     {
 3         #region Public Method
 4         /// <summary>
 5         /// 
 6         /// </summary>
 7         /// <param name="p"></param>
 8         /// <param name="s"></param>
 9         protected void ExecuteProtocolCommand(CommunicateProtocol p, SocketState s)
10         {
11             if (p == null) throw new ArgumentNullException("CommunicateProtocol is null");
12             switch (p.Command)
13             {
14                 case 0: CommandWrapper(((ICommandFunc)new Command0()), p, s); break;
15                 case 1: CommandWrapper(((ICommandFunc)new Command1()), p, s); break;
16                 case 2: CommandWrapper(((ICommandFunc)new Command2()), p, s); break;
17                 case 998: CommandWrapper(((ICommandFunc)new Command998()), p, s); break;
18                 case 999: CommandWrapper(((ICommandFunc)new Command999()), p, s); break;
19                 // 
20                 case 1001: CommandWrapper(((ICommandFunc)new Command1001()), p, s); break;
21                 case 1002: CommandWrapper(((ICommandFunc)new Command1002()), p, s); break;
22                 //
23                 case 2001: CommandWrapper(((ICommandFunc)new Command2001()), p, s); break;
24                 case 2002: CommandWrapper(((ICommandFunc)new Command2002()), p, s); break;
25                 //
26                 case 3001: CommandWrapper(((ICommandFunc)new Command3001()), p, s); break;
27 
28                 default: throw new Exception("Protocol type does not exist.");
29             }
30         }
31         /// <summary>
32         /// 
33         /// </summary>
34         /// <param name="func"></param>
35         /// <param name="p"></param>
36         /// <param name="s"></param>
37         protected abstract void CommandWrapper(ICommandFunc func, CommunicateProtocol p, SocketState s);
38         #endregion Public Method
39     }

View Code

以及在“命令”中封装，下一个命令。

 1     /// <summary>
 2     /// 
 3     /// </summary>
 4     public class Command1002 : ICommandFunc
 5     {
 6         #region ICommandFunc 成员
 7         public CommandProfile profile { get; set; }
 8 
 9         /// <summary>
10         /// 
11         /// </summary>
12         /// <param name="protocol"></param>
13         /// <param name="state"></param>
14         /// <param name="sobj"></param>
15         /// <returns></returns>
16         public SocketState Execute(CommunicateProtocol protocol, SocketState state, IHSSocket sobj)
17         {
18             state.IsReceiveThreadAlive = false;
19 
20             // Check File
21             if (!FileHelper.IsFileExist(profile.UpLoadPath + profile.UpLoadFName))
22             {
23                 var p = ProtocolMgmt.InitProtocolHeader(998, System.Text.Encoding.UTF8.GetBytes("服务端文件不存在"), null);
24                 ProtocolMgmt.SendProtocol(state, p, sobj);
25                 state.OutPutMsg = string.Format("Command 1002 ：服务端文件不存在");
26                 return state;
27             }
28             if (!FileHelper.CanRead(profile.UpLoadPath + profile.UpLoadFName))
29             {
30                 var p = ProtocolMgmt.InitProtocolHeader(998, System.Text.Encoding.UTF8.GetBytes("文件已被打开或占用，请稍后重试"), null);
31                 ProtocolMgmt.SendProtocol(state, p, sobj);
32                 state.OutPutMsg = string.Format("Command 1002 ：文件已被打开或占用");
33                 return state;
34             }
35 
36             FileInfo fi = new FileInfo(profile.UpLoadPath + profile.UpLoadFName);
37             using (FileStream fs = new FileStream(profile.UpLoadPath + profile.UpLoadFName, FileMode.Open))
38             {
39                 var p = ProtocolMgmt.InitProtocolHeader(2002, System.Text.Encoding.UTF8.GetBytes(fi.Name), fs);
40                 ProtocolMgmt.SendProtocol(state, p, sobj);
41                 state.OutPutMsg = string.Format("Command 1002 ：发送文件 {0} 成功。", fi.FullName);
42             }
43 
44             return state;
45         }
46 
47 
48         #endregion
49     }