如何通道输入的优先级的方式来处理?有什么等价 到Scala的“reactWithin(0) { ... case TIMEOUT }”建设?

有帮助吗?

解决方案

我写订阅类上的一组间隔开优先消息。这不是一个理想的情况下,一般的方式来消耗优先的消息,但我会后留给后人。我想自定义RequestReplyChannel将是某些其他情况下,更好的选择。 PriorityQueue中的实现是留给读者作为练习给读者。

class PrioritySubscriber<T> : BaseSubscription<T>
{
    private readonly PriorityQueue<T> queue;
    private readonly IScheduler scheduler;
    private readonly Action<T> receive;
    private readonly int interval;

    private readonly object sync = new object();
    private ITimerControl next = null;

    public PrioritySubscriber(IComparer<T> comparer, IScheduler scheduler,
        Action<T> receive, int interval)
    {
        this.queue = new PriorityQueue<T>(comparer);
        this.scheduler = scheduler;
        this.receive = receive;
        this.interval = interval;
    }

    protected override void OnMessageOnProducerThread(T msg)
    {
        lock (this.sync)
        {
            this.queue.Enqueue(msg);

            if (this.next == null)
            {
                this.next =
                    this.scheduler.Schedule(this.Receive, this.interval);
            }
        }
    }

    private void Receive()
    {
        T msg;

        lock (this.sync)
        {
            msg = this.queue.Dequeue();

            if (this.queue.Count > 0)
            {
                this.next =
                    this.scheduler.Schedule(this.Receive, this.interval);
            }
        }

        this.receive(msg);
    }
}
许可以下: CC-BY-SA归因
不隶属于 StackOverflow
scroll top