所以我解决这个问题的方法是将基本组件移动到它们自己的抽象对象中。然后建立在这些对象之上。例如，服务器只需要接受/跟踪连接。所以我会创建一个看起来像这样的服务器对象：
namespace MultiServerExample.Base
{
public interface IAsyncServerBase
{
void StartListening();
bool IsListening { get; }
void StopListening();
void WriteDataToAllClients(byte[] data);
}
public abstract class AsyncServerBase<TClientBase> : IAsyncServerBase
where TClientBase : IAsyncClientBase, new()
{
// implement a TcpListener to gain access to Active property
private sealed class ActiveTcpListener : TcpListener
{
public ActiveTcpListener(IPAddress localaddr, int port)
: base(localaddr, port) { }
public bool IsActive => Active;
}
// our listener object
private ActiveTcpListener Listener { get; }
// our clients
private ConcurrentDictionary<string, TClientBase> Clients { get; }
// construct with a port
public AsyncServerBase(int port)
{
Clients = new ConcurrentDictionary<string, TClientBase>();
Listener = new ActiveTcpListener(IPAddress.Any, port);
}
// virtual methods for client action
public virtual void OnClientConnected(TClientBase client) { }
public virtual void OnClientDisconnected(TClientBase client, Exception ex) { }
// start the server
public void StartListening()
{
if(!IsListening)
{
Listener.Start();
Listener.BeginAcceptTcpClient(OnAcceptedTcpClient, this);
}
}
// check if the server is running
public bool IsListening =>
Listener.IsActive;
// stop the server
public void StopListening()
{
if (IsListening)
{
Listener.Stop();
Parallel.ForEach(Clients, x => x.Value.DetachClient(null));
Clients.Clear();
}
}
// async callback for when a client wants to connect
private static void OnAcceptedTcpClient(IAsyncResult res)
{
var me = (AsyncServerBase<TClientBase>)res.AsyncState;
if (!me.IsListening) { return; }
try
{
TcpClient client = null;
try
{
client = me.Listener.EndAcceptTcpClient(res);
}
catch(Exception ex)
{
System.Diagnostics.Debug.WriteLine($"Warning: unable to accept client:\n{ex}");
}
if(client != null)
{
// create a new client
var t = new TClientBase();
// set up error callbacks
t.Error += me.OnClientBaseError;
// notify client we have attached
t.AttachClient(client);
// track the client
me.Clients[t.Id] = t;
// notify we have a new connection
me.OnClientConnected(t);
}
}
finally
{
// if we are still listening, wait for another connection
if(me.IsListening)
{
me.Listener.BeginAcceptSocket(OnAcceptedTcpClient, me);
}
}
}
// Event callback from a client that an error has occurred
private void OnClientBaseError(object sender, AsyncClientBaseErrorEventArgs e)
{
var client = (TClientBase)sender;
client.Error -= OnClientBaseError;
OnClientDisconnected(client, e.Exception);
client.DetachClient(e.Exception);
Clients.TryRemove(client.Id, out _);
}
// utility method to write data to all clients connected
public void WriteDataToAllClients(byte[] data)
{
Parallel.ForEach(Clients, x => x.Value.WriteData(data));
}
}
}
至此，运行服务器的所有基础知识都已经讲完了。现在对于在服务器上运行的客户端：
namespace MultiServerExample.Base
{
public interface IAsyncClientBase
{
event EventHandler<AsyncClientBaseErrorEventArgs> Error;
void AttachClient(TcpClient client);
void WriteData(byte[] data);
void DetachClient(Exception ex);
string Id { get; }
}
public abstract class AsyncClientBase : IAsyncClientBase
{
protected virtual int ReceiveBufferSize { get; } = 1024;
private TcpClient Client { get; set; }
private byte[] ReceiveBuffer { get; set; }
public event EventHandler<AsyncClientBaseErrorEventArgs> Error;
public string Id { get; }
public AsyncClientBase()
{
Id = Guid.NewGuid().ToString();
}
public void AttachClient(TcpClient client)
{
if(ReceiveBuffer != null) { throw new InvalidOperationException(); }
ReceiveBuffer = new byte[ReceiveBufferSize];
Client = client;
try
{
Client.GetStream().
BeginRead(ReceiveBuffer, 0, ReceiveBufferSize, OnDataReceived, this);
OnAttachedToServer();
}
catch (Exception ex)
{
Error?.Invoke(this,
new AsyncClientBaseErrorEventArgs(ex, "BeginRead"));
}
}
public void DetachClient(Exception ex)
{
try
{
Client.Close();
OnDetachedFromServer(ex);
}
catch { /* intentionally swallow */ }
Client = null;
ReceiveBuffer = null;
}
public virtual void OnDataReceived(byte[] buffer) { }
public virtual void OnAttachedToServer() { }
public virtual void OnDetachedFromServer(Exception ex) { }
public void WriteData(byte[] data)
{
try
{
Client.GetStream().BeginWrite(data, 0, data.Length, OnDataWrote, this);
}
catch(Exception ex)
{
Error?.Invoke(this, new AsyncClientBaseErrorEventArgs(ex, "BeginWrite"));
}
}
private static void OnDataReceived(IAsyncResult iar)
{
var me = (AsyncClientBase)iar.AsyncState;
if(me.Client == null) { return; }
try
{
var bytesRead = me.Client.GetStream().EndRead(iar);
var buf = new byte[bytesRead];
Array.Copy(me.ReceiveBuffer, buf, bytesRead);
me.OnDataReceived(buf);
}
catch (Exception ex)
{
me.Error?.Invoke(me, new AsyncClientBaseErrorEventArgs(ex, "EndRead"));
}
}
private static void OnDataWrote(IAsyncResult iar)
{
var me = (AsyncClientBase)iar.AsyncState;
try
{
me.Client.GetStream().EndWrite(iar);
}
catch(Exception ex)
{
me.Error?.Invoke(me,
new AsyncClientBaseErrorEventArgs(ex, "EndWrite"));
}
}
}
}
现在您的所有基本代码都已编写完毕。您无需以任何方式更改此设置。您只需实现自己的客户端和服务器即可做出相应的响应。例如，这是一个基本的服务器实现：
public class MyServer : AsyncServerBase<MyClient>
{
public MyServer(int port) : base(port)
{
}
public override void OnClientConnected(MyClient client)
{
Console.WriteLine($"* MyClient connected with Id: {client.Id}");
base.OnClientConnected(client);
}
public override void OnClientDisconnected(MyClient client, Exception ex)
{
Console.WriteLine($"***** MyClient disconnected with Id: {client.Id} ({ex.Message})");
base.OnClientDisconnected(client, ex);
}
}
这是上面服务器用于通信的客户端：
public class MyClient : AsyncClientBase
{
public override void OnAttachedToServer()
{
base.OnAttachedToServer();
Console.WriteLine($"{Id}: {GetType().Name} attached. Waiting for data...");
}
public override void OnDataReceived(byte[] buffer)
{
base.OnDataReceived(buffer);
Console.WriteLine($"{Id}: {GetType().Name} recieved {buffer.Length} bytes. Writing 5 bytes back.");
WriteData(new byte[] { 1, 2, 3, 4, 5 });
}
public override void OnDetachedFromServer(Exception ex)
{
base.OnDetachedFromServer(ex);
Console.WriteLine($"{Id}: {GetType().Name} detached.");
}
}
为了强调这一点，这里有另一个客户端，它可以简单地插入相同的服务器实现，但赋予它不同的特性：
public class MyOtherClient : AsyncClientBase
{
public override void OnAttachedToServer()
{
base.OnAttachedToServer();
Console.WriteLine($"{Id}: {GetType().Name} attached. Writing 4 bytes back.");
WriteData(new byte[] { 1, 2, 3, 4 });
}
public override void OnDataReceived(byte[] buffer)
{
base.OnDataReceived(buffer);
Console.WriteLine($"{Id}: {GetType().Name} recieved {buffer.Length} bytes.");
}
public override void OnDetachedFromServer(Exception ex)
{
base.OnDetachedFromServer(ex);
Console.WriteLine($"{Id}: {GetType().Name} detached.");
}
}
就使用它而言，这里有一个小测试程序，可以对其进行压力测试：
class Program
{
static void Main(string[] args)
{
var servers = new IAsyncServerBase[]
{
new MyServer(50000),
new MyServer(50001),
new MyOtherServer(50002)
};
foreach (var s in servers)
{
s.StartListening();
}
RunTestUsingMyServer("1", 89, 50000);
RunTestUsingMyServer("2", 127, 50001);
RunTestUsingMyOtherServer("3", 88, 50002);
Console.Write("Press any key to exit... ");
Console.ReadKey(true);
foreach (var s in servers)
{
s.WriteDataToAllClients(new byte[] { 1, 2, 3, 4, 5 });
s.StopListening();
}
}
private static void RunTestUsingMyServer(string name, int clientCount, int port)
{
Parallel.For(0, clientCount, x =>
{
using (var t = new TcpClient())
{
t.Connect(IPAddress.Loopback, port);
t.GetStream().Write(new byte[] { 1, 2, 3, 4, 5 }, 0, 5);
t.GetStream().Read(new byte[512], 0, 512);
t.Close();
}
Console.WriteLine($"FINISHED PASS {name} #{x}");
});
}
private static void RunTestUsingMyOtherServer(string name, int clientCount, int port)
{
Parallel.For(0, clientCount, x =>
{
using (var t = new TcpClient())
{
t.Connect(IPAddress.Loopback, port);
t.GetStream().Read(new byte[512], 0, 512);
t.GetStream().Write(new byte[] { 1, 2, 3, 4, 5, 6 }, 0, 6);
t.Close();
}
Console.WriteLine($"FINISHED PASS {name} #{x}");
});
}
}
如果有兴趣，这里是full source code，您可以查看。希望这能让您达到与重用代码有关的目标。