C# TCP非同步通訊框架封裝
阿新 • • 發佈：2019-02-10
最近用c#寫了一個TCP非同步通訊框架TCPHelper，用於服務端客戶端通訊，採用非同步和事件驅動的方式，使用者只需要初始化和裝載事件即可使用，框架圖粗略如下所示：
使用如下：（本文框架及例項下載地址）
服務端
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net.Sockets;
using System.Net;
using System.Threading;
using TCPHelper;
namespace Socket非同步請求Server
{
class 客戶端：
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using TCPHelper;
using System.Net;
using System.Net.Sockets;
namespace Socket非同步請求Client
{
class Program
{
static void Main(string[] args)
{
ClientAsync client = new ClientAsync();
client.Completed += new Action<System.Net.Sockets.TcpClient, EnSocketAction>((c, enAction) =>
{
IPEndPoint iep = c.Client.RemoteEndPoint as IPEndPoint;
string key = string.Format("{0}:{1}", iep.Address.ToString(), iep.Port);
switch (enAction)
{
case EnSocketAction.Connect:
Console.WriteLine("已經與{0}建立連線",key);
break;
case EnSocketAction.SendMsg:
Console.WriteLine("{0}：向{1}傳送了一條訊息",DateTime.Now,key);
break;
case EnSocketAction.Close:
Console.WriteLine("服務端連線關閉");
break;
default:
break;
}
});
client.Received += new Action<string,string>((key,msg)=>
{
Console.WriteLine("{0}對我說：{1}",key,msg);
});
client.ConnectAsync(10001);
while (true)
{
string msg = Console.ReadLine();
client.SendAsync(msg);
}
}
}
}
上面所引用的TCPHelper為所用的框架，下載地址：點選這裡下載
框架的原始碼也貼出來：
ClientAsync.cs:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net.Sockets;
using System.Net;
using System.Threading;
namespace TCPHelper
{
public class ClientAsync
{
private TcpClient client;
/// <summary>
/// 客戶端連線完成、傳送完成、連線異常或者服務端關閉觸發的事件
/// </summary>
public event Action<TcpClient,EnSocketAction> Completed;
/// <summary>
/// 客戶端接收訊息觸發的事件
/// </summary>
public event Action<string,string> Received;
/// <summary>
/// 用於控制非同步接收訊息
/// </summary>
private ManualResetEvent doReceive = new ManualResetEvent(false);
//標識客戶端是否關閉
private bool isClose = false;
public ClientAsync()
{
client = new TcpClient();
}
/// <summary>
/// 非同步連線
/// </summary>
/// <param name="ip">要連線的伺服器的ip地址</param>
/// <param name="port">要連線的伺服器的埠</param>
public void ConnectAsync(string ip, int port)
{
IPAddress ipAddress = null;
try
{
ipAddress = IPAddress.Parse(ip);
}
catch (Exception)
{
throw new Exception("ip地址格式不正確，請使用正確的ip地址！");
}
client.BeginConnect(ipAddress, port,ConnectCallBack, client);
}
/// <summary>
/// 非同步連線，連線ip地址為127.0.0.1
/// </summary>
/// <param name="port">要連線服務端的埠</param>
public void ConnectAsync(int port)
{
ConnectAsync("127.0.0.1", port);
}
/// <summary>
/// 非同步接收訊息
/// </summary>
private void ReceiveAsync()
{
doReceive.Reset();
StateObject obj=new StateObject();
obj.Client=client;
client.Client.BeginReceive(obj.ListData, 0, obj.ListData.Length, SocketFlags.None, ReceiveCallBack, obj);
doReceive.WaitOne();
}
/// <summary>
/// 非同步傳送訊息
/// </summary>
/// <param name="msg"></param>
public void SendAsync(string msg)
{
byte[] listData=Encoding.UTF8.GetBytes(msg);
client.Client.BeginSend(listData, 0, listData.Length, SocketFlags.None, SendCallBack,client);
}
/// <summary>
/// 非同步連線的回撥函式
/// </summary>
/// <param name="ar"></param>
private void ConnectCallBack(IAsyncResult ar)
{
TcpClient client = ar.AsyncState as TcpClient;
client.EndConnect(ar);
OnComplete(client, EnSocketAction.Connect);
}
/// <summary>
/// 非同步接收訊息的回撥函式
/// </summary>
/// <param name="ar"></param>
private void ReceiveCallBack(IAsyncResult ar)
{
StateObject obj = ar.AsyncState as StateObject;
int count=-1;
try
{
count = obj.Client.Client.EndReceive(ar);
doReceive.Set();
}
catch (Exception)
{
//如果發生異常，說明客戶端失去連線，觸發關閉事件
Close();
OnComplete(obj.Client, EnSocketAction.Close);
}
if (count > 0)
{
string msg = Encoding.UTF8.GetString(obj.ListData, 0, count);
if (!string.IsNullOrEmpty(msg))
{
if (Received != null)
{
IPEndPoint iep = obj.Client.Client.RemoteEndPoint as IPEndPoint;
string key = string.Format("{0}:{1}", iep.Address, iep.Port);
Received(key,msg);
}
}
}
}
private void SendCallBack(IAsyncResult ar)
{
TcpClient client = ar.AsyncState as TcpClient;
try
{
client.Client.EndSend(ar);
OnComplete(client, EnSocketAction.SendMsg);
}
catch (Exception)
{
//如果發生異常，說明客戶端失去連線，觸發關閉事件
Close();
OnComplete(client, EnSocketAction.Close);
}
}
public virtual void OnComplete(TcpClient client, EnSocketAction enAction)
{
if (Completed != null)
Completed(client, enAction);
if (enAction == EnSocketAction.Connect)//建立連線後，開始接收資料
{
ThreadPool.QueueUserWorkItem(x =>
{
while (!isClose)
{
try
{
Thread.Sleep(20);
ReceiveAsync();
Thread.Sleep(20);
}
catch (Exception)
{
Close();
OnComplete(client, EnSocketAction.Close);
}
}
});
}
}
public void Close()
{
isClose = true;
}
}
}
ServerAsync.cs:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net;
using System.Net.Sockets;
using System.Threading;
namespace TCPHelper
{
public class ServerAsync
{
private TcpListener listener = null;
//用於控制非同步接受連線
private ManualResetEvent doConnect = new ManualResetEvent(false);
//用於控制非同步接收資料
private ManualResetEvent doReceive=new ManualResetEvent(false);
//標識服務端連線是否關閉
private bool isClose = false;
private Dictionary<string,TcpClient> listClient=new Dictionary<string,TcpClient>();
/// <summary>
/// 已建立連線的集合
/// key:ip:port
/// value:TcpClient
/// </summary>
public Dictionary<string,TcpClient> ListClient
{
get{return listClient;}
private set{listClient=value;}
}
/// <summary>
/// 連線、傳送、關閉事件
/// </summary>
public event Action<string,EnSocketAction> Completed;
/// <summary>
/// 接收到資料事件
/// </summary>
public event Action<string,string> Received;
public ServerAsync()
{
}
/// <summary>
/// 開始非同步監聽ip地址的埠
/// </summary>
/// <param name="ip"></param>
/// <param name="port"></param>
public void StartAsync(string ip, int port)
{
IPAddress ipAddress=null;
try
{
ipAddress=IPAddress.Parse(ip);
}
catch(Exception e)
{
throw e;
}
listener = new TcpListener(new IPEndPoint(ipAddress, port));
listener.Start();
ThreadPool.QueueUserWorkItem(x =>
{
while (!isClose)
{
doConnect.Reset();
listener.BeginAcceptTcpClient(AcceptCallBack, listener);
doConnect.WaitOne();
}
});
}
/// <summary>
/// 開始非同步監聽本機127.0.0.1的埠號
/// </summary>
/// <param name="port"></param>
public void StartAsync(int port)
{
StartAsync("127.0.0.1", port);
}
/// <summary>
/// 開始非同步傳送資料
/// </summary>
/// <param name="key">客戶端的ip地址和埠號</param>
/// <param name="msg">要傳送的內容</param>
public void SendAsync(string key, string msg)
{
if (!ListClient.ContainsKey(key))
{
throw new Exception("所用的socket不在字典中,請先連線！");
}
TcpClient client = ListClient[key];
byte[] listData=Encoding.UTF8.GetBytes(msg);
client.Client.BeginSend(listData, 0, listData.Length, SocketFlags.None, SendCallBack, client);
}
/// <summary>
/// 開始非同步接收資料
/// </summary>
/// <param name="key">要接收的客戶端的ip地址和埠號</param>
private void ReceiveAsync(string key)
{
doReceive.Reset();
if (ListClient.ContainsKey(key))
{
TcpClient client = ListClient[key];
//if (!client.Connected)
//{
// ListClient.Remove(key);
// OnComplete(key, EnSocketAction.Close);
// return;
//}
StateObject obj = new StateObject();
obj.Client = client;
try
{
client.Client.BeginReceive(obj.ListData, 0, obj.ListData.Length, SocketFlags.None, ReceiveCallBack, obj);
}
catch (Exception)
{
}
doReceive.WaitOne();
}
}
/// <summary>
/// 非同步接收連線的回撥函式
/// </summary>
/// <param name="ar"></param>
private void AcceptCallBack(IAsyncResult ar)
{
TcpListener l= ar.AsyncState as TcpListener;
TcpClient client = l.EndAcceptTcpClient(ar);
doConnect.Set();
IPEndPoint iep = client.Client.RemoteEndPoint as IPEndPoint;
string key = string.Format("{0}:{1}", iep.Address.ToString(), iep.Port);
if (!ListClient.ContainsKey(key))
{
ListClient.Add(key, client);
OnComplete(key, EnSocketAction.Connect);
}
}
/// <summary>
/// 非同步傳送資料的回撥函式
/// </summary>
/// <param name="ar"></param>
private void SendCallBack(IAsyncResult ar)
{
TcpClient client= ar.AsyncState as TcpClient;
IPEndPoint iep = client.Client.RemoteEndPoint as IPEndPoint;
string key = string.Format("{0}:{1}", iep.Address.ToString(), iep.Port);
if (Completed != null)
{
Completed(key, EnSocketAction.SendMsg);
}
}
/// <summary>
/// 非同步接收資料的回撥函式
/// </summary>
/// <param name="ar"></param>
private void ReceiveCallBack(IAsyncResult ar)
{
StateObject obj =ar.AsyncState as StateObject;
int count=-1;
try
{
count = obj.Client.Client.EndReceive(ar);
}
catch (Exception e)
{
if (!obj.Client.Client.Connected)
{
IPEndPoint iep = obj.Client.Client.RemoteEndPoint as IPEndPoint;
string key = string.Format("{0}:{1}", iep.Address.ToString(), iep.Port);
ListClient.Remove(key);
OnComplete(key, EnSocketAction.Close);
doReceive.Set();
return;
}
}
doReceive.Set();
if (count > 0)
{
string msg = Encoding.UTF8.GetString(obj.ListData, 0, count);
if (!string.IsNullOrEmpty(msg))
{
if (Received != null)
{
IPEndPoint iep = obj.Client.Client.RemoteEndPoint as IPEndPoint;
string key = string.Format("{0}:{1}", iep.Address.ToString(), iep.Port);
Received(key,msg);//觸發接收事件
}
}
}
}
public virtual void OnComplete(string key, EnSocketAction enAction)
{
if (Completed != null)
Completed(key, enAction);
if (enAction == EnSocketAction.Connect)//當連線建立時，則要一直接收
{
ThreadPool.QueueUserWorkItem(x =>
{
while (ListClient.ContainsKey(key)&&!isClose)
{
Thread.Sleep(20);
ReceiveAsync(key);
Thread.Sleep(20);
}
});
}
}
public void Close()
{
isClose=true;
}
}
}
其他：
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Text;
namespace TCPHelper
{
/// <summary>
/// 接收socket的行為
/// </summary>
public enum EnSocketAction
{
/// <summary>
/// socket發生連線
/// </summary>
Connect = 1,
/// <summary>
/// socket傳送資料
/// </summary>
SendMsg = 2,
/// <summary>
/// socket關閉
/// </summary>
Close = 4
}
/// <summary>
/// 對非同步接收時的物件狀態的封裝，將socket與接收到的資料封裝在一起
/// </summary>
public class StateObject
{
public TcpClient Client { get; set; }
private byte[] listData = new byte[2048];
/// <summary>
/// 接收的資料
/// </summary>
public byte[] ListData
{
get
{
return listData;
}
set
{
listData = value;
}
}
}
}
當然此框架目前只能傳送接收字串，待添加發送接收byte的功能。