首頁猿問 AsyncProducerCons...

AsyncProducerConsumerQueue 的可觀察包裝器

C#

慕尼黑8549860 2023-09-16 16:12:18

AsyncProducerConsumerQueue<T>因此，我使用以下代碼為 Stephen Cleary 創(chuàng)建了一個可觀察的包裝器。我想知道這里是否有人知道我如何以更簡單的方式做到這一點？是否可以在沒有包裝類的情況下編寫它？是否可以防止將多個包裝器應用于一個隊列時出現錯誤？我可以讓它在第一個訂閱上連接，而不是通過直接調用嗎Connect？如果是這樣，這意味著什么？最后，你會怎么做呢？using Nito.AsyncEx;using System.Reactive;static async Task ExampleUsage() { var queue = new AsyncProducerConsumerQueue<int>(); var observable = queue.AsConnectableObservable(); await queue.EnqueueAsync(1); observable.Subscribe(Console.WriteLine); observable.Connect(); await queue.EnqueueAsync(2);}public static class AsyncExExtensions { public static IConnectableObservable<T> AsConnectableObservable<T>(this AsyncProducerConsumerQueue<T> queue) { return new ConnectableObservableForAsyncProducerConsumerQueue<T>(queue); }}class ConnectableObservableForAsyncProducerConsumerQueue<T> : IConnectableObservable<T> { readonly AsyncProducerConsumerQueue<T> Queue; long _isConnected = 0; ImmutableList<IObserver<T>> Observers = ImmutableList<IObserver<T>>.Empty; public ConnectableObservableForAsyncProducerConsumerQueue(AsyncProducerConsumerQueue<T> queue) { Queue = queue; } public IDisposable Connect() { if (Interlocked.Exchange(ref _isConnected, 1) == 1) throw new InvalidOperationException("Observable cannot be connected more than once."); var cts = new CancellationTokenSource(); var token = cts.Token; Task.Run(async () => { try { while (true) { token.ThrowIfCancellationRequested(); var @event = await Queue.DequeueAsync(token).ConfigureAwait(false); foreach (var observer in Observers) observer.OnNext(@event); } } catch (Exception x) when (x is OperationCanceledException || x is InvalidOperationException) { foreach (var observer in Observers) observer.OnCompleted(); } });

查看完整描述

1 回答

holdtom

TA貢獻1805條經驗獲得超10個贊

考慮以下兩個演示。當您有多個觀察者時，這些行為會有所不同。在第一個演示中，觀察者將競爭隊列中的項目，在第二個演示中，他們每個人都會收到一個副本。

演示 #1 - 冷可觀察

var queue = new AsyncProducerConsumerQueue<int>();

// This is a cold observable, so each observer is fed by its own individual dequeue loop

// and therefore will be 'competing' with other observers for queued items.

var coldObservable = Observable

// Create an observable that asynchronously waits for items to become available on the

// queue and then emits them to the observer. This will be cancelled when the observer

// is unsubscribed.

.Create<int>(async (observer, cancellationToken) =>

{

while (true)

{

var item = await queue.DequeueAsync(cancellationToken).ConfigureAwait(false);

Console.WriteLine($"Dequeued {item}");

observer.OnNext(item);

}

})

// If an InvalidOperationException is thrown by the above, continue with

// an empty observable instead of the error. This effectively catches an

// `OnError(InvalidOperationException)` and turns it into an `OnCompleted()`.

.Catch<int, InvalidOperationException>(exn =>

{

Console.WriteLine("Caught InvalidOperation");

return Observable.Empty<int>();

});

Console.WriteLine("TEST COLD");

await queue.EnqueueAsync(1);

Console.WriteLine("Enqueued 1");

Console.WriteLine("Subscribing A");

coldObservable.Subscribe(

item => Console.WriteLine($"A received {item}"),

() => Console.WriteLine("A completed"));

Console.WriteLine("Subscribing B");

coldObservable.Subscribe(

item => Console.WriteLine($"B received {item}"),

() => Console.WriteLine("B completed"));

await queue.EnqueueAsync(2);

Console.WriteLine("Enqueued 2");

await queue.EnqueueAsync(3);

Console.WriteLine("Enqueued 3");

queue.CompleteAdding();

Console.WriteLine("Completed adding");

Console.WriteLine("Waiting...");

await Task.Delay(2000);

Console.WriteLine("DONE");

// TEST COLD

// Enqueued 1

// Subscribing A

// Dequeued 1

// A received 1

// Subscribing B

// Enqueued 2

// Enqueued 3

// Completed adding

// Waiting...

// Dequeued 2

// Dequeued 3

// A received 2

// B received 3

// Caught InvalidOperation

// A completed

// B completed

// DONE

演示 #2 - 熱可觀察

var queue = new AsyncProducerConsumerQueue<int>();

var coldObservable = // defined same as above

// This is a hot observable, so each observer receives the same items from the queue.

var hotObservable = coldObservable

// Publish the cold observable to create an `IConnectableObservable` that will subscribe

// to the dequeue loop when connected and emit the same items to all observers.

.Publish()

// Automatically connect to the published observable when the first observer subscribes

// and automatically disconnect when the last observer unsubscribes. This means that the

// first observer will receive any items queued before it subscribes, but additional

// observers will only receive items queued after they subscribed.

.RefCount();

Console.WriteLine("TEST HOT");

await queue.EnqueueAsync(1);

Console.WriteLine("Enqueued 1");

Console.WriteLine("Subscribing A");

hotObservable.Subscribe(

item => Console.WriteLine($"A received {item}"),

() => Console.WriteLine("A completed"));

Console.WriteLine("Subscribing B");

hotObservable.Subscribe(

item => Console.WriteLine($"B received {item}"),

() => Console.WriteLine("B completed"));

await queue.EnqueueAsync(2);

Console.WriteLine("Enqueued 2");

await queue.EnqueueAsync(3);

Console.WriteLine("Enqueued 3");

queue.CompleteAdding();

Console.WriteLine("Completed adding");

Console.WriteLine("Waiting...");

await Task.Delay(2000);

Console.WriteLine("DONE");

// TEST HOT

// Enqueued 1

// Subscribing A

// Dequeued 1

// A received 1

// Subscribing B

// Enqueued 2

// Enqueued 3

// Dequeued 2

// Completed adding

// Waiting...

// A received 2

// B received 2

// Dequeued 3

// A received 3

// B received 3

// Caught InvalidOperation

// A completed

// B completed

// DONE

回答你原來的問題：

是否可以在沒有包裝類的情況下編寫它？

是的，請參閱上面的演示。

是否可以防止將多個包裝器應用于一個隊列時出現錯誤？

上面演示的方法不會阻止其他方將項目出隊（或在隊列上執(zhí)行任何其他操作）。如果您想確保只公開給定隊列的單個隊列IObservable<T>，請考慮通過創(chuàng)建一個ObservableProducerConsumerQueue<T>在內部創(chuàng)建和管理自己的AsyncProducerConsumerQueue. 您可以公開一個EnqueueAsync僅委托給內部隊列的方法，并使用上面演示的可觀察量之一將可觀察量公開為屬性或實現接口IObservable<T>。

我可以讓它在第一個訂閱上連接，而不是通過直接調用 Connect 嗎？如果是這樣，這意味著什么？

演示 #2 顯示了此行為并描述了其含義。如果您希望能夠在連接之前訂閱觀察者，請?zhí)^調用RefCount并像以前一樣使用IConnectableObservable返回的值Publish。

最后，你會怎么做呢？

如上所述，我將封裝隊列并使用上面演示的方法之一公開IObservable或公開。IConnectableObservable

反對回復 2023-09-16