强制 Observable 订阅者一次只写入一个流

Question

我目前正在使用 observables 来管理总线上生成的消息，这些消息被推送到各种流中。

一切正常，但由于消息可以进入，系统可能会尝试一次将多条消息写入流（即来自多个线程的消息），或者消息的发布速度比写入速度快流...如您所见，这会在写入时引起问题。

因此，我试图弄清楚如何组织事物，以便在收到消息时一次只处理一个。有什么想法吗？

public class MessageStreamResource : IResourceStartup
{
    private readonly IBus _bus;
    private readonly ISubject<string> _sender;

    public MessageStreamResource(IBus bus)
    {
        _bus = bus;

        _senderSubject = new Subject<string>();

        //`All` can publish messages at the same time as it's
        //collecting data being generated from different threads
        _bus.All.Subscribe(message => Observable.Start(() => ProcessMessage(message), TaskPoolScheduler.Default));

        //Note the above hops off the calls context so that the 
        //writing to the stream wont slow down the caller.
    }

    public void Configure(IAppBuilder app)
    {
        app.Map("/stream", async context =>
        {
            ...

            await context.Response.WriteAsync("Lets party!\n");
            await context.Response.Body.FlushAsync();

            var unSubscribe = _sender.Subscribe(async t =>
            {
                //PROBLEM HERE
                //I only want this callback to be executed 
                //one at a time...

                await context.Response.WriteAsync($"{t}\n");
                await context.Response.Body.FlushAsync();
            });

            ...

            await HoldOpenTask;
        });
    }

    private void ProcessMessage(IMessage message)
    {
        _sender.OnNext(message.Payload);
    }
}

Answer 1

如果我正确理解了这个问题，这可能可以通过SemaphoreSlim 来完成：

// ...
var semaphore = new SemaphoreSlim(initialCount: 1);

var unSubscribe = _sender.Subscribe(async t =>
{
    //PROBLEM HERE
    //I only want this callback to be executed 
    //one at a time...

    await semaphore.WaitAsync();
    try
    {
        await context.Response.WriteAsync($"{t}\n");
        await context.Response.Body.FlushAsync();
    }
    finally
    {
        semaphore.Release();
    }
});

SemaphoreSlim 是 IDisposable，请务必在适当的时候将其丢弃。

更新，第二次看，MapExtensions.Map 接受Action<IAppBuilder>，所以你传递了一个async void lambda，本质上创建了一堆即发即弃的异步操作。 Map 呼叫将返回给呼叫者，而他们可能仍在徘徊。这很可能不是您想要的，是吗？