c# - 使用 C# 的新异步功能等待网络数据包的最佳方式是什么

Question

我最近一直在尝试使用新的 Async CTP，但遇到了一种情况，我不确定如何继续。

在我当前的代码库中，我使用了“作业”和“作业管理器”的概念。作业的存在仅仅是为了处理初始消息、发送响应，然后等待响应。

我已经有了基于同步套接字的现有代码，其中网络线程正在等待数据到达，然后将其传递给事件处理程序，并最终传递给作业管理器。

作业管理器查找将处理该消息的作业，并将其传递。

所以场景是这样的:

1. 工作经理收到新消息并启动工作。
2. 作业启动，处理消息，并发送回复消息。
3. 此时作业将等待对回复的响应。

这是一个伪代码示例:

class MyJob : Job
{
    public override void RunJob( IPacketMsg packet )
    {
        // handle packet

        var myReply = new Packet();
        SendReply( myReply );

        await GetResponse();
    }
}

但我不完全确定如何在第 3 步继续进行。作业管理器将获得响应，然后将其传递给正在运行的作业。但我不确定如何让作业等待响应。

我考虑过创建一个只阻塞在 WaitHandle 上的等待任务，但这是最好的解决方案吗？

在这种情况下我还能做些什么吗？

编辑关于异步 CTP 的主题，在未使用 UI 的情况下会发生什么。我已经阅读了 Eric Lippert 的 Async 博客，但我认为它从未涉及过在没有 UI 线程的情况下一切如何在后台工作的主题(它是从后台 worker 中分离出来还是......？)

Answer 1

1. Job manager gets a new message and launches a job.
2. The job starts, processes the message, and sends a reply message.
3. At this point the job would wait for a response to the reply.

首先，我应该提到 Async CTP 可以很好地处理异步操作，但异步事件处理得不是很好。您可能需要考虑基于 Rx 的方法。但让我们暂时使用异步 CTP。

创建任务有两个基本选项:

• 与代表。例如，Task.Factory.StartNew 将在线程池上运行委托(delegate)。自定义任务工厂和调度程序为您提供更多任务委托(delegate)选项(例如，指定委托(delegate)必须在 STA 线程上运行)。
• 没有代表。例如，TaskFactory.FromAsync 包装现有的 Begin/End 方法对，TaskEx.FromResult 返回一个“ future 常量” "，并且 TaskCompletionSource 可用于显式控制 Task(FromAsync 和 FromResult 使用 TCS 内部)。

如果作业处理受 CPU 限制，则将其传递给 Task.Factory.StartNew 是有意义的。我将假设作业处理受 CPU 限制。

作业管理器伪代码:

// Responds to a new message by starting a new job on the thread pool.
private void RespondToNewMessage(IPacketMsg message)
{
  IJob job = ..;
  Task.Factory.StartNew(job.RunJob(message));
}

// Holds tasks waiting for a response.
private ConcurrentDictionary<int, TaskCompletionSource<IResponse>> responseTasks = ..;

// Asynchronously gets a response for the specified reply.
public Task<IResponse> GetResponseForReplyAsync(int replyId)
{
  var tcs = new TaskCompletionSource<IResponse>();
  responseTasks.Add(replyId, tcs);
  return tcs.Task;
}

// Responds to a new response by completing and removing its task.
private void RespondToResponse(IResponse response)
{
  var tcs = responseTasks[response.ReplyId];
  responseTasks.Remove(response.ReplyId);
  tcs.TrySetComplete(response);
}

想法是作业管理器还管理未完成响应的列表。为了实现这一点，我引入了一个简单的 int 回复标识符，作业管理器可以使用它来确定哪个响应对应哪个回复。

现在工作可以像这样工作:

public override void RunJob(IPacketMsg packet)
{
  // handle packet
  var myReply = new Packet();
  var response = jobManager.GetResponseForReplyAsync(myReply.ReplyId);
  SendReply(myReply);

  await response;
}

由于我们将作业放在线程池线程上，所以有一些棘手的事情:

1. GetResponseForReplyAsync 必须在发送回复之前调用(注册任务)，然后再await。这是为了避免在我们有机会注册之前发送回复并收到回复的情况。
2. RespondToResponse 将在完成之前删除任务注册，以防万一完成任务导致使用相同 ID 发送另一个回复。

如果作业足够短，不需要放在线程池线程上，那么解决方案就可以简化。