确保以正确的顺序发送排队的消息

Question

我有一个客户端-服务器架构，使用 TCP 进行通信。客户端以 1msg/70ms 的速率向客户端发送消息。服务器必须模拟网络延迟，因此在服务器接收端，我执行以下操作：

//Event handler for new messages
private void DataReceived(object sender, DataReceivedEventArgs args)
{
    //Add raw data to a queue for processing at a later time
    TemporaryRawMessages.Enqueue(args.Bytes);
    args.Recycle();
}

//Function running on thread; processes newly arrived raw messages
//new messages are removed from the queue, given their own thread and processed independently. 
void ProcessMessagesIn()
{
    while (true)
    {

        if (TemporaryRawMessages.Count > 0)
        {
            var raw = TemporaryRawMessages.Dequeue();
            var t = new Thread(() => {
                Thread.Sleep(LatencyUp);
                var message = (ClientToServerMessage)Utils.Deserialize(raw_message);
                Messages_In.Enqueue(message);
            });
            t.Start();
        }

    }
}

//Create & start a thread to processes incoming messages
var processMessagesIn = new Thread(ProcessMessagesIn);
processMessagesIn.Start();

注意LatencyUp 变量。我希望来自客户端的消息延迟 b x 毫秒，但平均间隔为 1msg/70ms。换句话说，所有消息相隔 70 毫秒，但偏移 LatencyUp。

我还为从服务器到客户端的消息添加了延迟，以表示网络停机延迟：

Task.Factory.StartNew(() => { SendLoop(); });

void SendToClient(ServerToClientMessage message)
{
    Messages_Out.Enqueue(message);
}

void SendLoop()
{
    while (true)
    {
        if(Messages_Out.Count > 0)
        {
            ServerToClientMessage message_out = null;
            Messages_Out.TryDequeue(out message_out);

            if(message_out != null)
            {
                Thread myNewThread = new Thread(() =>
                {
                    Thread.Sleep(LatencyDown);
                    if (ClientConnection != null && ClientConnection.State == ConnectionState.Connected)
                    {
                        var serialized = Utils.Serialize(message);
                        ClientConnection.SendBytes(serialized, SendOption.Reliable);
                    }
                });

                myNewThread.Start();
            }
        }
    }
}

我也试过这样发送，没有从服务器到客户端的消息队列：

void SendToClient(ServerToClientMessage message)
{
    Thread myNewThread = new Thread(() =>
    {
        Thread.Sleep(LatencyDown);
        Send(message);
    });

    myNewThread.Start();
}

问题

有时从服务器到客户端的消息会混淆...例如，消息 2 可能在消息 1 之前发送。

我不完全确定为什么。消息以正确的顺序放入队列（我已经检查过了）。在添加到队列和出队 + 发送之间的某个地方，出了点问题。

我唯一能想到的是一个线程在另一个线程之前完成，但我想不出这是怎么发生的。

我可以做些什么来确保消息按照发送的顺序（客户端到服务器）被接收并以正确的顺序从服务器发送到客户端？

性能（在处理速度方面）很重要，因此对于该主题的任何改进/建议也将不胜感激。

Answer 1

您不能使用多个线程按顺序分派事件。如果你想为你的数据生产者提供一个异步消费者，你只需要一个线程。

创建生产者-消费者队列的最简单方法是使用BlockingCollection<T>：

var blockingCollection = new BlockingCollection<int>();

// this is a single, one and only producer thread
Task.Run(() =>
{
    var number = 0;
    while (true)
    {
        number++;
        Console.WriteLine("tx --> " + number);
        blockingCollection.Add(number);
        Task.Delay(1000).Wait();
    }
});

// single, one and only consumer thread
foreach (var item in blockingCollection.GetConsumingEnumerable())
{
    Task.Delay(200).Wait();
    Console.WriteLine("    " + item + " --> rx");
}