在任务中使用异步套接字

Question

我有一个应用程序正在使用 .net 4.0 中的任务处理 FIFO 队列中的项目。

我是 .net 中的 TPL 和 Tasks 的新手，想知道是否有一个简单的解决方案可以解决我的问题：

Task 中的 Action 委托被分配给在异步套接字上发送和接收数据的方法。我遇到的问题是任务“过早地”结束。如何告诉任务等到所有通信完成后再处理队列中的下一个项目？

一种解决方案是切换到使用同步套接字，但我希望有一种方法可以使用异步套接字来做到这一点。

编辑 添加了一些代码：

class Program
{
    private BlockingCollection<string> myQueue;
    private CancellationTokenSource cancellationSignalForConsumeTask;
    private CancellationTokenSource cancellationSignalForProcessCommandTask;
    private AsyncSocket mySocket;

    public void Main(string[] args)
   {
       mySocket = new mySocket();
       myscoket.ReceiveData += mySocket_ReceiveData;

       cancellationSignalForConsumeTask = new CancellationTokenSource();
       Task listenerTask = Task.Factory.StartNew((obj) => Consume(),
                                                          cancellationSignalForConsumeTask.Token,
                                                          TaskCreationOptions.LongRunning);

        while (true)
        {}
   }

    private void Consume()
    {
       while (!myQueue.IsCompleted )
            {
                string _item = myQueue.Take();

                cancellationSignalForProcessCommandTask = new CancellationTokenSource();
                Task t = new Task(() => 
                    {
                        cancellationSignalForProcessCommandTask.Token.ThrowIfCancellationRequested();
                        DoSomeWork(_item);
                    }, cancellationSignalForProcessCommandTask.Token, TaskCreationOptions.LongRunning);

                t.Start();
                t.Wait();
            }
    }

    private void DoSomeWork(string _item)
    {
       mySocket.SendData("Data to server that could take a long time to process")
    }

    private void mySocket_ReceiveData(string dataFromServer)
    {
        string returnMessage = dataFromServer;

        //I want the Task to end here...

    }
}

问题是任务在DoSomeWork() 方法完成时结束（我明白为什么），有没有办法可以手动告诉任务通过CancellationTokenSource 对象结束？

Answer 1

如果我理解正确，您希望等待接收数据的任务，但您当前正在等待发送数据的任务。一种方法是使用类似AutoResetEvent 的构造：

private AutoResetEvent autoResetEvent = new AutoResetEvent(false);

private void Consume()
{
   while (!myQueue.IsCompleted )
        {
            string _item = myQueue.Take();

            cancellationSignalForProcessCommandTask = new CancellationTokenSource();
            Task t = new Task(() => 
                {
                    cancellationSignalForProcessCommandTask.Token.ThrowIfCancellationRequested();
                    DoSomeWork(_item);
                }, cancellationSignalForProcessCommandTask.Token, TaskCreationOptions.LongRunning);

            t.Start();

            // Wait for data to be received.
            // This line will block until autoResetEvent.Set() is called.
            autoResetEvent.WaitOne();
        }
}

private void mySocket_ReceiveData(string dataFromServer)
{
    string returnMessage = dataFromServer;

   // Notify other threads that data was received and that processing can continue.
   autoResetEvent.Set();
}

这只是使用 AutoResetEvent 的一个示例 - 您可能希望对其进行改进以满足您的需要。