ConcurrentBag.ToObservable() 运行一次并过早完成

Question

我有一个静态集合，比如调用远程休息 api 的任务：

static ConcurrentBag<Task<HttpResponseMessage>> _collection = new ConcurrentBag<Task<HttpResponseMessage>>();

static void Main(string[] args)
{
    Task.Factory.StartNew(() => Produce());
    Task.Factory.StartNew(() => Consume());

    Console.ReadKey();
}

一个线程向其中添加新项目：

private static void Produce()
{
    while (true)
    {
        var task = HttpClientFactory.Create().GetAsync("http://example.com");
        _collection.Add(task);
        Thread.Sleep(500);
    }
}

另一个线程应该处理这些项目：

private static void Consume()
{
    _collection.ToObservable()
               .Subscribe(
                   t => Console.WriteLine("++"),
                   ex => Console.WriteLine(ex.Message),
                   () => Console.WriteLine("Done"));
}

但它只运行一次并且过早完成。所以输出是;

++

完成

Answer 1

如果它像那样工作会很有趣......但遗憾的是它没有。 ToObservable 扩展方法是在 IEnumerable<T> 接口上定义的 - 所以它会获取集合的时间点快照。

您需要一个可以观察到的集合，例如ObservableCollection。有了这个，您可以响应添加事件以提供 Rx 管道（可能通过将CollectionChanged 事件与Observable.FromEventPattern 连接起来）。请记住，此集合不支持并发添加。这种技术是“进入 monad”的一种方式（即获得IObservable<T>）。

等效是将您的请求负载添加到主题。无论哪种方式，您都可以将它们投影到异步请求中。所以说（为了争论），你的 Produce 签名看起来像这样：

private static async Task<HttpResponseMessage> Produce(string requestUrl)

然后你可以构造一个 observable 来使用你的 Produce 方法将 requestUrls 转换为异步 Web 请求，如下所示：

var requests = new Subject<string>();
var responses = requests.SelectMany(
    x => Observable.FromAsync(() => Produce(x)));

responses.Subscribe(
               t => Console.WriteLine("++"),
               ex => Console.WriteLine(ex.Message),
               () => Console.WriteLine("Done"));

并使用以下内容提交每个请求：

requests.OnNext("http://myurl");

如果您需要并发添加，请参阅Observable.Synchronize。

如果您需要控制处理响应的线程，请使用ObserveOn，我对here 进行了冗长的解释。