响应门控下一条消息发送，与 Rx

Question

给定一个 List，我用我的 Send(message) 发送第一条消息。现在我想在发送下一条消息之前等待（异步）响应返回...

阻止直到通知“旧”方式

我知道如何使用线程锁定/针对这种情况实现基于事件的解决方案使用 Monitor.Wait 和 Monitor.Pulse

反应式“新”方式？

但我想知道使用反应式扩展是否有意义 /a> 在这里？

如果 Rx 可以在这里传达有价值的好处，那么我如何才能使响应以反应方式控制下一个发送调用？显然它会涉及 IObservable，可能有两个作为主要来源，但是然后呢？

Answer 1

这个问题不是很具体，似乎很笼统，因为您没有提到发送者接收者是什么等，所以答案也很笼统:)

var receiveObs = //You have created a observable around the receive mechanism 
var responses = messages.Select(m => {
   send(m);
   return receiveObs.First();
}).ToList();

Answer 2

我认为 Rx 是一个不错的选择，但我认为我可能会遗漏您的要求中的某些内容。据我了解，Rx 提供了一个非常简单的解决方案。

如果您已经有一个消息列表，那么您可以像这样响应式发送它们：

messages
    .ToObservable()
    .ObserveOn(Scheduler.ThreadPool)
    .Subscribe(m =>
    {
        Send(m);
    });

这会将对 Send 的调用推送到线程池，并且通过可观察对象的内置行为，每次调用 < code>Send 等待上一个调用完成。

由于这一切都发生在不同的线程上，因此您的代码是非阻塞的。

Rx 的额外好处是您无需更改 Send 方法的行为或签名即可实现此目的。

很简单吧？

我对此进行了测试，鉴于我对您的问题的理解，它工作得很好。这是您所需要的全部还是我错过了什么？

Answer 3

我不确定 Rx 是否适合这里。 Rx 基于“推送集合”的概念，即将数据推送给消费者而不是拉取数据。您想要的是拉取第一个项目，异步发送它，并在异步操作完成时继续下一个元素。对于这种工作，完美的工具是async / await*！

async void SendMessages(List<Message> messages)
{
    foreach (Message message in messages)
    {
        await SendAsync(message);
    }
}

Task SendAsync(Message message);

* 在异步 CTP 或 .NET 4.5 预览版中可用

Answer 4

假设您的 Send 方法遵循 APM 模型，则以下方法应该适合您

List<Message> messages;
IObservable<Response> xs;
xs =  messages.ToObservable().SelectMany(msg => Observable.FromAsyncPattern(Send, msg));

编辑 - 这不会像安德森建议的那样起作用，这里有一个显示问题的示例

Func<int,string> Send = (ii) => { "in Send".Dump(); Thread.Sleep(2000); return ii.ToString(); };
Func<int,IObservable<string>> sendIO =  Observable.FromAsyncPattern<int,string>(Send.BeginInvoke, Send.EndInvoke);
(new [] { 1, 2, 3 }).ToObservable().SelectMany(sendIO).Dump();