最近我不得不构建一个小型 TCP 客户端应用程序，它连接到外部应用程序的 TCP 侦听器，旨在处理大量数据量和高频率。
我围绕 TCPClient 类创建了一个包装类，只是为了捕获异常，并保持对一些感兴趣的属性（网络流等）的引用。这是包装：
public class MyTCPClient
{
private string serverIP;
private int serverPort;
public TcpClient tcpClient = new TcpClient();
private IPEndPoint serverEndPoint;
private NetworkStream stream = null;
public string name;
public MyTCPClient(string serverIp, int serverPort, string parentName)
{
this.serverIP = serverIp;
this.serverPort = serverPort;
this.name = parentName + "_TCPClient";
serverEndPoint = new IPEndPoint(IPAddress.Parse(serverIP), serverPort);
tcpClient.ReceiveBufferSize = 1048576;
this.TryConnect();
}
private bool TryConnect()
{
try
{
tcpClient.Connect(serverEndPoint);
}
catch (SocketException e1)
{
throw new ErrorOnConnectingException(e1, "SocketException while connecting. (see msdn Remarks section for more details. ) Error code: " + e1.ErrorCode);
}
catch (ArgumentNullException e2)
{
throw new ErrorOnConnectingException(e2, "ArgumentNullException while connecting. (The hostname parameter is null.) Message: " + e2.Message);
}
catch (ArgumentOutOfRangeException e3)
{
throw new ErrorOnConnectingException(e3, "ArgumentOutOfRangeException while connecting (The port parameter is not between MinPort and MaxPort. ). Message: " + e3.Message);
}
catch (ObjectDisposedException e4)
{
throw new ErrorOnConnectingException(e4, "ObjectDisposedException while connecting. (TcpClient is closed. ) Message: " + e4.Message);
}
try
{
stream = this.tcpClient.GetStream();
}
catch (ObjectDisposedException e1)
{
throw new ErrorOnGettingStreamException(e1, "ObjectDisposedException while acquiring Network stream. (The TcpClient has been closed. ) Message: " + e1.Message);
}
catch (InvalidOperationException e2)
{
throw new ErrorOnGettingStreamException(e2, "ArgumentOutOfRangeException while acquiring Network stream (The TcpClient is not connected to a remote host. ). Message: " + e2.Message);
}
return true;
}
public string ReadData()
{
try
{
ASCIIEncoding encoder = new ASCIIEncoding();
byte[] dataHeader = new byte[12];
if (this.tcpClient.Connected)
{
stream.Read(dataHeader, 0, 12);
}
else
{
throw new ErrorOnReadingException(null, "The underlying TCP tcpClient is not connected any more");
}
var strHeaderMessage = System.Text.Encoding.Default.GetString(dataHeader);
Utils.logToTimeStampedFile(strHeaderMessage, name);
int bodyAndTailCount = Convert.ToInt32(strHeaderMessage.Replace("#", ""));
byte[] dataBodyAndTail = new byte[bodyAndTailCount];
if (this.tcpClient.Connected)
{
stream.Read(dataBodyAndTail, 0, bodyAndTailCount);
}
else
{
throw new ErrorOnReadingException(null, "The underlying TCP tcpClient is not connected any more");
}
var strBodyAndTailMessage = System.Text.Encoding.Default.GetString(dataBodyAndTail);
Utils.logToTimeStampedFile(strBodyAndTailMessage, name);
return strBodyAndTailMessage;
}
catch (FormatException e0)
{
CloseAllLeft();
throw new ErrorOnReadingException(e0, "FormatException while reading data. (Bytes red are null or does not correspond to specification, happens on closing Server) Message: " + e0.Message);
}
catch (ArgumentNullException e1)
{
CloseAllLeft();
throw new ErrorOnReadingException(e1, "ArgumentNullException while reading data. (The buffer parameter is null.) Message: " + e1.Message);
}
catch (ArgumentOutOfRangeException e2)
{
CloseAllLeft();
throw new ErrorOnReadingException(e2, "ArgumentOutOfRangeException while reading data. (see msdn description) Message: " + e2.Message);
}
catch (IOException e3)
{
CloseAllLeft();
throw new ErrorOnReadingException(e3, "IOException while reading data. (The underlying Socket is closed.) Message: " + e3.Message);
}
catch (ObjectDisposedException e4)
{
CloseAllLeft();
throw new ErrorOnReadingException(e4, "ArgumentOutOfRangeException while reading data. (see msdn description) Message: " + e4.Message);
}
}
public void CloseAllLeft()
{
try
{
stream.Close();
}
catch (Exception e)
{
Console.WriteLine("Exception closing tcp network stream: " + e.Message);
}
try
{
tcpClient.Close();
}
catch (Exception e)
{
Console.WriteLine("Exception closing tcpClient: " + e.Message);
}
}
}
尽管如此，没有提到使用这个 MyTCPClient 的线程。该应用程序应该有两个这样的 TCP 客户端，在不同的端口上连接，并执行不同的工作。我是 TCP 编程的新手，在对属性进行了一些游荡之后，我决定使用阻塞读取方法——即默认情况下 TCPClient.Read() 方法将阻塞线程，直到有新数据。我需要这种方法，因为我无法控制外部应用程序的侦听器，并且识别服务器关闭的唯一方法是根据 TCP 套接字规范发送“零字节”。
因此，我构建了一个抽象类，它将维护和控制稍后将使用上述 MyTCPClient 类的线程（根据设计，最终可能会阻塞父线程）。这是我的抽象 TCPManager 的代码：
/// <summary>
/// Serves as a dispatcher for the high frequency readings from the TCP pipe.
/// Each time the thread is started it initializes new TCPClients which will attempt to connect to server.
/// Once established a TCP socket connection is alive until the thread is not requested to stop.
///
/// Error hanling level here:
///
/// Resources lke NetworkStream and TCPClients are ensured to be closed already within the myTCPClient class, and the error handling here
/// is steps on top of that - sending proper emails, notifications and logging.
///
/// </summary>
public abstract class AbstractmyTCPClientManager
{
public string name;
public string serverIP;
public int serverPort;
public Boolean requestStop = false;
public Boolean MyTCPClientThreadRunning = false;
public Boolean requestStart = false;
public myTCPClient myTCPClient;
public int sleepInterval;
public Thread MyTCPClientThread;
public AbstractmyTCPClientManager(string name, string serverIP, int serverPort)
{
this.name = name;
this.serverIP = serverIP;
this.serverPort = serverPort;
}
public void ThreadRun()
{
MyTCPClientThreadRunning = false;
bool TCPSocketConnected = false;
bool AdditionalInitializationOK = false;
// keep trying to init requested tcp clients
while (!MyTCPClientThreadRunning && !requestStop) // and we are not suggested to stop
{
while (!TCPSocketConnected && !requestStop) // and we are not suggested to stop)
{
try
{
myTCPClient = new myTCPClient(serverIP, serverPort, name);
TCPSocketConnected = true;
}
catch (ErrorOnConnectingException e0)
{
// nah, too long message
string detail = e0.originalException != null ? e0.originalException.Message : "No inner exception";
//Utils.logToTimeStampedFile("Creating connection attempt failed.(1." + e0.customMessage + " 2." + detail + "). Will retry in 10 seconds...", name);
//Utils.logToTimeStampedFile(e0.customMessage + " (" + detail + "). Will retry in 10 seconds...", name);
Utils.logToTimeStampedFile(detail + ". Will retry in 10 seconds...", name);
Thread.Sleep(10000);
}
catch (ErrorOnGettingStreamException e1)
{
// nah, too long message
string detail = e1.originalException != null ? e1.originalException.Message : "No inner exception";
//Utils.logToTimeStampedFile("Getting network stream attempt failed. (1." + e1.customMessage + " 2." + detail + "). Will retry in 10 seconds...", name);
//Utils.logToTimeStampedFile(e1.customMessage + " (" + detail + "). Will retry in 10 seconds...", name);
Utils.logToTimeStampedFile(detail + ". Will retry in 10 seconds...", name);
Thread.Sleep(10000);
}
}
Utils.logToTimeStampedFile("TCP Communication established", name);
while (!AdditionalInitializationOK && !requestStop) // or we are not suggested to stop
{
try
{
AdditionalInitialization();
AdditionalInitializationOK = true;
}
catch (AdditionalInitializationException e1)
{
string detail = e1.originalException != null ? e1.originalException.Message : "No inner exception";
//Utils.logToTimeStampedFile("Additional initialization failed (1." + e1.customMessage + " 2." + detail + "). Will retry in 10 seconds", name);
Utils.logToTimeStampedFile(e1.customMessage + ". Will retry in 10 seconds", name);
Thread.Sleep(10000);
}
}
MyTCPClientThreadRunning = TCPSocketConnected && AdditionalInitializationOK;
ViewModelLocator.ControlTabStatic.updateUIButtons();
}
Utils.logToTimeStampedFile("Additional Initialization successfully completed, thread started", name);
// while all normal (i.e nobody request a stop) continiously sync with server (read data)
while (!requestStop)
{
try
{
syncWithInterface();
}
catch (ErrorOnReadingException e1)
{
string detail = e1.originalException != null ? e1.originalException.Message : "No inner exception";
//Utils.logToTimeStampedFile("Error ocured while reading data. (1." + e1.customMessage + " 2." + detail + ")", name);
Utils.logToTimeStampedFile(e1.customMessage, name);
if (!requestStop) // i.e if this indeed is an exception, during a normal flow, and nobody requested a thread stop (which migh cause read exceptions as a consequence)
{
Utils.logToTimeStampedFile("There was no external stop request, when the error occured, doing tcp client restart.", name);
requestStop = true;
requestStart = true;
}
}
Thread.Sleep(sleepInterval);
}
// we need to close all after execution, but the execution may be closed before/while resources were still initializing
if (TCPSocketConnected)
{
myTCPClient.CloseAllLeft();
}
if (AdditionalInitializationOK)
{
ReleaseAdditionalResources();
}
// remember that thread is stoped
MyTCPClientThreadRunning = false;
Utils.logToTimeStampedFile("Thread stoped", name);
ViewModelLocator.ControlTabStatic.updateUIButtons();
// this serves as a restart
if (requestStart)
{
Utils.logToTimeStampedFile("Restarting thread...", name);
this.requestStop = false;
this.requestStart = false; // we are already processing a request start event, so reset this flag
this.MyTCPClientThread = new Thread(new ThreadStart(this.ThreadRun));
this.MyTCPClientThread.Name = this.name;
this.MyTCPClientThread.IsBackground = true;
this.MyTCPClientThread.Start();
}
}
/// <summary>
/// this method empties the entire TCP buffer, cycling through it
/// </summary>
private void syncWithInterface()
{
int counter = 0;
// read at most 100 messages at once (we assume that for 3 sec interval there might not be more,
//even if they are, it is still OK, they just will be processed next time)
while (counter < 100)
{
counter++;
string data = myTCPClient.ReadData();
ForwardData(data);
}
// below is left for testing:
/*
* "Sleep(0) or Yield is occasionally useful in production code for
* advanced performance tweaks. It’s also an excellent diagnostic tool
* for helping to uncover thread safety issues: if inserting Thread.Yield()
* anywhere in your code makes or breaks the program, you almost certainly have a bug."*/
Thread.Yield();
}
/// <summary>
/// Left for implementing in the caller that initialized the object. Meaning: one and the same way for receiving market/order data. Different ways of processing this data
/// </summary>
/// <param name="data"></param>
public abstract void ForwardData(string data);
/// <summary>
/// left for implementing in child classes. Its purpose is to initialize any additional resources needed for the thread to operate.
/// If something goes wrong while getting this additional resources,
/// an AdditionalInitialization exception should be thrown, which is than handled from the initialization phase in the caller.
/// </summary>
public abstract void AdditionalInitialization();
// countrapart of AdditionalInitialization method - what is initialized should be then closed
public abstract void ReleaseAdditionalResources();
}
之后，每个需要的 TCP 通信通道都会有上述抽象类的专用实现，提供 ForwardData（即如何处理此数据）和 AdditionalInitialization（即在特定 TCP 通信处理之前还需要初始化什么）方法的实现正在运行。例如，我的线程中的一个需要在接收数据之前初始化额外的存储线程）。
一切都很好，除了关闭 TCP 处理。我有这个 requestStop 变量来控制线程是否应该退出或继续，但问题是 Read() 方法可能会陷入连续阻塞，甚至阻止 requestStop 变量被读取（我应该说我需要的两个 tcp 通道过程非常不同，其中一个非常频繁地接收数据，而另一个 - 偶尔接收）。我仍然希望他们实现相同的设计。因此，从我目前所阅读的内容来看，我必须实现另一个“父”或“控制”或“包装”线程，它实际上将在观察 requestStop 参数时完成工作。
我正在寻找像这篇文章这样的解决方案，或者像这篇文章这样的计时器
任何建议将不胜感激。谢谢！