中,我们编写了客户端功能。
这一篇将讲解ISocketHandler的实现。
再来回顾一下ISocketHandler接口。
public interface ISocketHandler{ ////// 开始接收 /// /// Socket网络流 /// 回调函数 /// 自定义状态 ///异步结果 IAsyncResult BeginReceive(Stream stream, AsyncCallback callback, object state); ////// 结束接收 /// /// 异步结果 ///接收到的数据 byte[] EndReceive(IAsyncResult asyncResult); ////// 开始发送 /// /// 要发送的数据 /// 数据偏移 /// 发送长度 /// Socket网络流 /// 回调函数 /// 自定义状态 ///异步结果 IAsyncResult BeginSend(byte[] data, int offset, int count, Stream stream, AsyncCallback callback, object state); ////// 结束发送 /// /// 异步结果 ///发送是否成功 bool EndSend(IAsyncResult asyncResult);}
做一个类SocketHandler继承ISocketHandler接口
////// Socket处理程序/// public class SocketHandler : ISocketHandler{ ////// 开始接收 /// /// Socket网络流 /// 回调函数 /// 自定义状态 ///异步结果 public IAsyncResult BeginReceive(Stream stream, AsyncCallback callback, object state) { } ////// 结束接收 /// /// 异步结果 ///接收到的数据 public byte[] EndReceive(IAsyncResult asyncResult) { } ////// 开始发送 /// /// 要发送的数据 /// 数据偏移 /// 发送长度 /// Socket网络流 /// 回调函数 /// 自定义状态 ///异步结果 public IAsyncResult BeginSend(byte[] data, int offset, int count, Stream stream, AsyncCallback callback, object state) { } ////// 结束发送 /// /// 异步结果 ///发送是否成功 public bool EndSend(IAsyncResult asyncResult) { }}
增加两个属性与构造函数。
//异步处理关系集合 private DictionaryStateSet; //发送队列 private List SendQueue; /// /// 实例化Socket处理程序 /// public SocketHandler() { StateSet = new Dictionary(); SendQueue = new List (); }
StateSet可以保存我们的异步调用结果等数据
SendQueue用来做一个发送队列
接下来我们从发送数据开始。
由于需要用到Stream的异步方法,我们需要定义一个State类。
internal class SocketHandlerState{ ////// 数据 /// public byte[] Data { get; set; } ////// 异步结果 /// public IAsyncResult AsyncResult { get; set; } ////// Socket网络流 /// public Stream Stream { get; set; } ////// 异步回调函数 /// public AsyncCallback AsyncCallBack { get; set; } ////// 是否完成 /// public bool Completed { get; set; } ////// 数据长度 /// public int DataLength { get; set; }}
因为我们需要返回IAsyncResult,所以我们继承该接口做一个SocketAsyncResult类。
////// Socket异步操作状态/// public class SocketAsyncResult : IAsyncResult{ ////// 实例化Socket异步操作状态 /// /// public SocketAsyncResult(object state) { AsyncState = state; AsyncWaitHandle = new AutoResetEvent(false); } ////// 获取用户定义的对象,它限定或包含关于异步操作的信息。 /// public object AsyncState { get; private set; } ////// 获取用于等待异步操作完成的 System.Threading.WaitHandle。 /// public WaitHandle AsyncWaitHandle { get; private set; } ////// 获取一个值,该值指示异步操作是否同步完成。 /// public bool CompletedSynchronously { get { return false; } } ////// 获取一个值,该值指示异步操作是否已完成。 /// public bool IsCompleted { get; internal set; }}
然后开始编写发送数据相关函数。
这里我将发送数据的大小限制为最大65535。
只需发送长度为2的头信息即可把数据长度发送到对方。
////// 开始发送 /// /// 要发送的数据 /// 数据偏移 /// 发送长度 /// Socket网络流 /// 回调函数 /// 自定义状态 ///异步结果 public IAsyncResult BeginSend(byte[] data, int offset, int count, Stream stream, AsyncCallback callback, object state) { //data不能为null if (data == null) throw new ArgumentNullException("data"); //offset不能小于0和超过data长度 if (offset > data.Length || offset < 0) throw new ArgumentOutOfRangeException("offset"); //count不能大于65535 if (count <= 0 || count > data.Length - offset || count > ushort.MaxValue) throw new ArgumentOutOfRangeException("count"); //stream不能为null if (stream == null) throw new ArgumentNullException("stream"); //回调函数不能为null if (callback == null) throw new ArgumentNullException("callback"); //stream异常 if (!stream.CanWrite) throw new ArgumentException("stream不支持写入。"); SocketAsyncResult result = new SocketAsyncResult(state); //初始化SocketHandlerState SocketHandlerState shs = new SocketHandlerState(); shs.Data = data; shs.AsyncResult = result; shs.Stream = stream; shs.AsyncCallBack = callback; shs.DataLength = 0; //锁定SendQueue //避免多线程同时发送数据 lock (SendQueue) { //添加状态 SendQueue.Add(shs); //如果SendQueue数量大于1,则表示有数据尚未发送完成 if (SendQueue.Count > 1) return result; } //获取数据长度 //ushort的最大值为65535 //转换为byte[]长度为2 var dataLength = BitConverter.GetBytes((ushort)data.Length); //向对方发送长度为2的头信息,表示接下来要发送的数据长度 stream.Write(dataLength, 0, dataLength.Length); //开始异步发送数据 stream.BeginWrite(shs.Data, 0, shs.Data.Length, EndWrite, shs).AsyncWaitHandle.WaitOne(); return result; } //stream异步结束写入 private void EndWrite(IAsyncResult ar) { SocketHandlerState state = (SocketHandlerState)ar.AsyncState; //锁定StateSet lock (StateSet) StateSet.Add(state.AsyncResult, state); try { state.Stream.EndWrite(ar); } catch { //出现Socket异常,发送失败 state.Completed = false; //允许等待线程继续 ((AutoResetEvent)state.AsyncResult.AsyncWaitHandle).Set(); //执行异步回调函数 state.AsyncCallBack(state.AsyncResult); return; } //发送成功 state.Completed = true; //允许等待线程继续 ((AutoResetEvent)state.AsyncResult.AsyncWaitHandle).Set(); //执行异步回调函数 state.AsyncCallBack(state.AsyncResult); //锁定SendQueue lock (SendQueue) { SocketHandlerState prepare = null; //移除当前发送完成的数据 SendQueue.Remove(state); //如果SendQueue还有数据存在,则继续发送 if (SendQueue.Count > 0) { prepare = SendQueue[0]; } if (prepare != null) { //获取数据长度 //ushort的最大值为65535 //转换为byte[]长度为2 var dataLength = BitConverter.GetBytes((ushort)prepare.Data.Length); //向对方发送长度为2的头信息,表示接下来要发送的数据长度 prepare.Stream.Write(dataLength, 0, dataLength.Length); //开始异步发送数据 prepare.Stream.BeginWrite(prepare.Data, 0, prepare.Data.Length, EndWrite, prepare).AsyncWaitHandle.WaitOne(); } } } ////// 结束发送 /// /// 异步结果 ///发送是否成功 public bool EndSend(IAsyncResult asyncResult) { //判断异步操作状态是否属于当前处理程序 if (!StateSet.ContainsKey(asyncResult)) throw new ArgumentException("无法识别的asyncResult。"); SocketHandlerState state = StateSet[asyncResult]; lock (StateSet) StateSet.Remove(asyncResult); return state.Completed; }
接下来是接收数据的相关方法。
////// 开始接收 /// /// Socket网络流 /// 回调函数 /// 自定义状态 ///异步结果 public IAsyncResult BeginReceive(Stream stream, AsyncCallback callback, object state) { //stream不能为null if (stream == null) throw new ArgumentNullException("stream"); //回调函数不能为null if (callback == null) throw new ArgumentNullException("callback"); //stream异常 if (!stream.CanRead) throw new ArgumentException("stream不支持读取。"); SocketAsyncResult result = new SocketAsyncResult(state); //初始化SocketHandlerState SocketHandlerState shs = new SocketHandlerState(); shs.Data = new byte[2]; shs.AsyncResult = result; shs.Stream = stream; shs.AsyncCallBack = callback; shs.Completed = true; //开始异步接收长度为2的头信息 //该头信息包含要接收的主要数据长度 stream.BeginRead(shs.Data, 0, 2, EndRead, shs); return result; } //stream异步结束读取 private void EndRead(IAsyncResult ar) { SocketHandlerState state = (SocketHandlerState)ar.AsyncState; int dataLength; try { dataLength = state.Stream.EndRead(ar); } catch { dataLength = 0; } //dataLength为0则表示Socket断开连接 if (dataLength == 0) { lock (StateSet) StateSet.Add(state.AsyncResult, state); //设定接收到的数据位空byte数组 state.Data = new byte[0]; //允许等待线程继续 ((AutoResetEvent)state.AsyncResult.AsyncWaitHandle).Set(); //执行异步回调函数 state.AsyncCallBack(state.AsyncResult); return; } //如果是已完成状态,则表示state.Data的数据是头信息 if (state.Completed) { //设定状态为未完成 state.Completed = false; //已接收得数据长度为0 state.DataLength = 0; //获取主要数据长度 var length = BitConverter.ToUInt16(state.Data, 0); //初始化数据的byte数组 state.Data = new byte[length]; try { //开始异步接收主要数据 state.Stream.BeginRead(state.Data, 0, length, EndRead, state); } catch { //出现Socket异常 lock (StateSet) StateSet.Add(state.AsyncResult, state); state.Data = new byte[0]; ((AutoResetEvent)state.AsyncResult.AsyncWaitHandle).Set(); state.AsyncCallBack(state.AsyncResult); } return; } //接收到主要数据 else { //判断是否接收了完整的数据 if (dataLength + state.DataLength != state.Data.Length) { //增加已接收数据长度 state.DataLength += dataLength; try { //继续接收数据 state.Stream.BeginRead(state.Data, state.DataLength, state.Data.Length - state.DataLength, EndRead, state); } catch { //出现Socket异常 lock (StateSet) StateSet.Add(state.AsyncResult, state); state.Data = new byte[0]; ((AutoResetEvent)state.AsyncResult.AsyncWaitHandle).Set(); state.AsyncCallBack(state.AsyncResult); return; } return; } //接收完成 state.Completed = true; lock (StateSet) StateSet.Add(state.AsyncResult, state); ((AutoResetEvent)state.AsyncResult.AsyncWaitHandle).Set(); state.AsyncCallBack(state.AsyncResult); } } ////// 结束接收 /// /// 异步结果 ///接收到的数据 public byte[] EndReceive(IAsyncResult asyncResult) { //判断异步操作状态是否属于当前处理程序 if (!StateSet.ContainsKey(asyncResult)) throw new ArgumentException("无法识别的asyncResult。"); SocketHandlerState state = StateSet[asyncResult]; lock (StateSet) StateSet.Remove(asyncResult); return state.Data; }
至此,SocketHandler的功能已经实现。
将为大家讲解服务器端的实现。
原文地址: