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

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的编码方法是什么，这样在对HandleNext()的多个并发调用中就不会发生MarkAsHandledBy()和IsHandledBy()之间的竞争？

编辑：

这是为了进行长时间的民意调查。对每个web请求调用CCD_ 4。该请求只能处理一条消息，然后返回到客户端。下一个请求接收下一条消息，依此类推。

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

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));
}

在CCD_ 13中去掉CCD_；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定义了不需要并发通知的流的概念。

也就是说，为了回答您的问题，为了避免出现Race条件，请在IsReceivedBy和MarkAsReceivedFor方法中加一个锁。

至于更好的方法，您可以放弃整个处理业务，在接收请求时使用ConcurrentQueue和TryDequeue消息（您只执行Take(1)，这符合队列模型）。Rx可以帮助您为每条消息指定一个TTL并将其从队列中删除，但您也可以对传入请求执行此操作。