如何在响应式扩展中避免这种竞争条件

Question

我在方法中有这样的代码：

ISubject<Message> messages = new ReplaySubject<Message>(messageTimeout);

public void HandleNext(string clientId, Action<object> callback)
{
    messages.Where(message => !message.IsHandledBy(clientId))
            .Take(1)
            .Subscribe(message =>
                       {
                           callback(message.Message);
                           message.MarkAsHandledBy(clientId);
                       });
}

编码它的 rx'y 方法是什么，这样MarkAsHandledBy() 和 IsHandledBy() 之间就不会在多个并发调用 HandleNext() 时发生竞争？

编辑：

这是用于长轮询。 HandleNext() 为每个 Web 请求调用。请求只能处理一条消息，然后返回给客户端。下一个请求接收下一条消息，依此类推。

完整的代码（当然仍在进行中）是这样的：

public class Queue
{
    readonly ISubject<MessageWrapper> messages;

    public Queue() : this(TimeSpan.FromSeconds(30)) {}

    public Queue(TimeSpan messageTimeout)
    {
        messages = new ReplaySubject<MessageWrapper>(messageTimeout);
    }

    public void Send(string channel, object message)
    {
        messages.OnNext(new MessageWrapper(new List<string> {channel}, message));
    }

    public void ReceiveNext(string clientId, string channel, Action<object> callback)
    {
        messages
            .Where(message => message.Channels.Contains(channel) && !message.IsReceivedBy(clientId))
            .Take(1)
            .Subscribe(message =>
                       {
                           callback(message.Message);
                           message.MarkAsReceivedFor(clientId);
                       });
    }

    class MessageWrapper
    {
        readonly List<string> receivers;

        public MessageWrapper(List<string> channels, object message)
        {
            receivers = new List<string>();
            Channels = channels;
            Message = message;
        }

        public List<string> Channels { get; private set; }
        public object Message { get; private set; }

        public void MarkAsReceivedFor(string clientId)
        {
            receivers.Add(clientId);
        }

        public bool IsReceivedBy(string clientId)
        {
            return receivers.Contains(clientId);
        }
    }
}

编辑 2：

现在我的代码如下所示：

public void ReceiveNext(string clientId, string channel, Action<object> callback)
{
    var subscription = Disposable.Empty;
    subscription = messages
        .Where(message => message.Channels.Contains(channel))
        .Subscribe(message =>
                   {
                       if (message.TryDispatchTo(clientId, callback))
                           subscription.Dispose();
                   });
}

class MessageWrapper
{
    readonly object message;
    readonly List<string> receivers;

    public MessageWrapper(List<string> channels, object message)
    {
        this.message = message;
        receivers = new List<string>();
        Channels = channels;
    }

    public List<string> Channels { get; private set; }

    public bool TryDispatchTo(string clientId, Action<object> handler)
    {
        lock (receivers)
        {
            if (IsReceivedBy(clientId)) return false;
            handler(message);
            MarkAsReceivedFor(clientId);
            return true;
        }
    }

    void MarkAsReceivedFor(string clientId)
    {
        receivers.Add(clientId);
    }

    bool IsReceivedBy(string clientId)
    {
        return receivers.Contains(clientId);
    }
}

Answer 1

在我看来，您正在为自己制造 Rx 噩梦。 Rx 应该提供一种非常简单的方法来将订阅者连接到您的消息。

我喜欢这样一个事实，即您有一个自包含的类来保存您的 ReplaySubject - 它可以阻止您代码中的其他地方恶意并过早调用 OnCompleted。

但是，ReceiveNext 方法不提供任何方法让您删除订阅者。至少是内存泄漏。您在 MessageWrapper 中跟踪客户端 ID 也是潜在的内存泄漏。

我建议你尝试使用这种函数而不是ReceiveNext:

public IDisposable RegisterChannel(string channel, Action<object> callback)
{
    return messages
        .Where(message => message.Channels.Contains(channel))
        .Subscribe(message => callback(message.Message));
}

这是非常 Rx-ish。这是一个很好的纯查询，您可以轻松取消订阅。

由于Action<object> callback 无疑与clientId 直接相关，我会考虑在其中放置防止重复消息处理的逻辑。

现在你的代码是非常程序化的，不适合 Rx。似乎您还没有完全了解如何最好地使用 Rx。这是一个好的开始，但您需要更多地从功能上思考（如在函数式编程中）。

---

如果您必须按原样使用您的代码，我建议您进行一些更改。

在Queue 中这样做：

public IDisposable ReceiveNext(
    string clientId, string channel, Action<object> callback)
{
    return
        messages
            .Where(message => message.Channels.Contains(channel))
            .Take(1)
            .Subscribe(message =>
                message.TryReceive(clientId, callback));
}

在MessageWrapper 中去掉MarkAsReceivedFor 和IsReceivedBy 并改为这样做：

    public bool TryReceive(string clientId, Action<object> callback)
    {
        lock (receivers)
        {
            if (!receivers.Contains(clientId))
            {
                callback(this.Message);
                receivers.Add(clientId);
                return true;
            }
            else
                return false;
        }
    }

虽然我真的不明白为什么你有 .Take(1)，但这些更改可能会减少竞争条件，具体取决于其原因。

Answer 2

我不确定像这样使用 Rx 是一个好习惯。 Rx 定义了流的概念，它要求没有并发通知。

也就是说，要回答您的问题，为了避免竞争条件，请在 IsReceivedBy 和 MarkAsReceivedFor 方法中加锁。

至于更好的方法，您可以放弃整个处理业务，在收到请求时使用ConcurrentQueue 和TryDequeue 消息（您只是在做Take(1) - 这适合队列模型）。 Rx 可以帮助您给每条消息一个 TTL 并将其从队列中删除，但您也可以在传入请求时这样做。