TcpListener 每 5 秒发送一次心跳并从客户端异步读取消息

Question

我正在编写一个 TcpListener 服务，它作为 Windows 服务运行，供应商软件客户端将连接到该服务。我无法更改客户的实施，必须遵守他们的规范。

我有接收消息并返回响应的部分正常工作。我遇到的问题是客户希望每 5 秒发送一次心跳消息 (S0000E)，它会回复相同的消息。我不确定如何在我正在执行的异步/等待内容中添加该功能，以处理从客户端收到的真实消息。

开机

_serverListenerTask = Task.Run(() => AcceptClientsAsync(_listener, _cancellationToken.Token));

AcceptClientsAsync

static async Task AcceptClientsAsync(TcpListener listener, CancellationToken ct)
{
    var clientCounter = 0;
    while (!ct.IsCancellationRequested)
    {
        TcpClient client = await listener.AcceptTcpClientAsync()
                                         .ConfigureAwait(false);
        clientCounter++;
        await ReceiveMessageAsync(client, clientCounter, ct);
    }
}

ReceiveMessageAsync

static async Task ReceiveMessageAsync(TcpClient client, int clientIndex, CancellationToken ct)
{
    Log.Info("New client ({0}) connected", clientIndex);
    using (client)
    {
        var buffer = new byte[4096];
        var stream = client.GetStream();
        while (!ct.IsCancellationRequested)
        {
            var timeoutTask = Task.Delay(TimeSpan.FromSeconds(15));
            var amountReadTask = stream.ReadAsync(buffer, 0, buffer.Length, ct);

            var completedTask = await Task.WhenAny(timeoutTask, amountReadTask)
                                          .ConfigureAwait(false);

            if (completedTask == timeoutTask)
            {
                var msg = Encoding.ASCII.GetBytes("Client timed out");
                await stream.WriteAsync(msg, 0, msg.Length);
                break;
            }

            var bytesRead = amountReadTask.Result;
            if (bytesRead == 0)
            {
                // Nothing was read
                break;
            }

            // Snip... Handle message from buffer here

            await stream.WriteAsync(responseBuffer, 0, responseBuffer.Length, ct)
                        .ConfigureAwait(false);
        }
    }
    Log.Info("Client ({0}) disconnected", clientIndex);
}

我以为我可以向Task.WhenAny 添加一个心跳任务，但这会导致心跳总是触发，我永远无法读取响应。我还尝试在超时和读取任务之前发送心跳，这对发送很有用，但是我要么读取心跳响应而不是下一条消息，要么超时任务将完成并断开客户端。本质上，如果心跳交换成功，那么客户端不应在 15 秒延迟后断开连接。

Answer 1

实现 TCP 服务器-客户端并不是一项简单的任务。但是，如果您对其进行改进以提高资源效率，则以下实施方式可能是一种实用的解决方案：

服务器：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Server
{
    public class Program
    {
        static List<SocketBuffer> clients = new List<SocketBuffer>();
        public static void Main(string[] args)
        {
            //Receive from any IP, listen on port 65000 in this machine
            var listener = new TcpListener(IPAddress.Any, 65000);
            var t = Task.Run(() =>
            {
                while (true)
                {
                    listener.Start();
                    var task = listener.AcceptTcpClientAsync();
                    task.Wait();
                    clients.Add(new SocketBuffer(task.Result, new byte[4096]));
                }
            });
            t.Wait();    //It will remain here, do in a better way if you like !
        }
    }

    /// <summary>
    /// We need this class because each TcpClient will have its own buffer
    /// </summary>
    class SocketBuffer
    {
        public SocketBuffer(TcpClient client, byte[] buffer)
        {
            this.client = client;
            stream = client.GetStream();
            this.buffer = buffer;

            receiveData(null);
        }

        private TcpClient client;
        private NetworkStream stream;
        private byte[] buffer;

        private object _lock = new object();
        private async void receiveData(Task<int> result)
        {
            if (result != null)
            {
                lock (_lock)
                {
                    int numberOfBytesRead = result.Result;
                    //If no data read, it means we are here to be notified that the tcp client has been disconnected
                    if (numberOfBytesRead == 0)
                    {
                        onDisconnected();
                        return;
                    }
                    //We need a part of this array, you can do it in more efficient way if you like
                    var segmentedArr = new ArraySegment<byte>(buffer, 0, numberOfBytesRead).ToArray();
                    OnDataReceived(segmentedArr);
                }

            }
            var task = stream.ReadAsync(buffer, 0, buffer.Length);
            //This is not recursion in any sense because the current 
            //thread will be free and the call to receiveData will be from a new thread
            await task.ContinueWith(receiveData);       
        }

        private void onDisconnected()
        {
            //Add your code here if you want this event
        }

        private void OnDataReceived(byte[] dat)
        {
            //Do anything with the data, you can reply here. I will just pring the received data from the demo client
            string receivedTxt = Encoding.ASCII.GetString(dat);
            Console.WriteLine(receivedTxt);
        }
    }
}

演示客户端：

using System;
using System.Net.Sockets;
using System.Text;
using System.Threading;

namespace Client
{
    public class Program
    {
        public static void Main(string[] args)
        {
            TcpClient client = new TcpClient();
            var task = client.ConnectAsync("localhost", 65000);
            task.Wait();
            if(client.Connected)
            {
                Console.WriteLine("Client connected");
                var stream = client.GetStream();
                var data = Encoding.ASCII.GetBytes("test");
                stream.Write(data, 0, data.Length);
            }
            else
            {
                Console.WriteLine("Client NOT connected");
            }
            Thread.Sleep(60000);
        }
    }
}