使用 Akka 进行 Scala 折叠

Question

我在 Java 中实现了我所谓的“可折叠队列”，即 ExecutorService 使用的 LinkedBlockingQueue。这个想法是每个任务作为一个唯一的 id，如果在队列中，而另一个任务通过相同的 id 提交，它不会被添加到队列中。 Java 代码如下所示：

public final class FoldablePricingQueue extends LinkedBlockingQueue<Runnable> {
    @Override
    public boolean offer(final Runnable runnable) {
        if (contains(runnable)) {
            return true; // rejected, but true not to throw an exception
        } else {
            return super.offer(runnable);
        }
    }
}

线程必须预先启动，但这是一个小细节。我有一个实现 Runnable 的抽象类，它需要一个唯一的 id ......这是传入的那个

我想使用 Scala 和 Akka（Actor）实现相同的逻辑。 我需要访问邮箱，我想我需要覆盖！方法并检查事件的邮箱..以前有人这样做过吗？

Answer 1

这正是 Akka 邮箱的工作原理。 Akka 邮箱在任务队列中只能存在一次。

看看：

https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/dispatch/Dispatcher.scala#L143

https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/dispatch/Dispatcher.scala#L198

使用原子布尔值非常便宜地实现，因此无需遍历队列。

另外，顺便说一下，Java 中的 Queue 已损坏，因为它没有覆盖 put、add 或 offer(E, long, TimeUnit)。

Answer 2

也许你可以用两个演员来做到这一点。一个facade 一个和一个worker 一个。客户将作业发送至facade。 Facade 然后转发到 worker，并在其内部状态中记住它们，一个 Set queuedJobs。当它收到queued 的作业时，它会丢弃它。每次worker 开始处理一个作业（或完成它，无论哪个适合你），它都会向 Facade 发送一条 StartingOn(job) 消息，这会将其从 queuedJobs 中删除。

Answer 3

建议的设计没有意义。最接近Runnable 的是Actor。当然，您可以将它们保存在列表中，如果它们已经存在，则不要添加它们。此类列表由 routing actors 保存，可以从 Akka 提供的现成部分创建，也可以使用 forward 方法从基本参与者创建。

你不能查看其他演员的邮箱，覆盖! 没有任何意义。您所做的是将所有消息发送到路由参与者，然后该路由参与者将它们转发到正确的目的地。

自然，由于它接收这些消息，它可以在那时执行任何逻辑。