使用Akka进行Scala折叠

Question

我在Java中实现了一个“可折叠队列”，即由ExecutorService使用的LinkedBlockingQueue。其思想是每个任务都有一个唯一的ID，如果该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（Actors）实现相同的逻辑。我需要访问邮箱，并且我认为我需要重写!方法并检查邮箱中的事件...有人以前做过这个吗？

- jts

---

刚刚发布了这篇文章，正在阅读：https://dev59.com/tk3Sa4cB1Zd3GeqPvHCo - jts

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中的队列是有问题的，因为它没有覆盖put、add或offer(E、long、TimeUnit)。

- Viktor Klang

---

抱歉，您误解了我的问题。我不是在问每个 Q 中 MB 实例的数量，而是关于如果它们已经在 MB 中，是否可以将任务接受到 MB 中。 - jts

最简单的方法可能就是为您的调度程序创建自己的邮箱实现。看看它是如何完成的，然后覆盖enqueue方法即可。 - Viktor Klang

重写调度程序的createMailbox方法，以返回一个执行去重操作的邮箱。请参考Akka 1.2版本中的ExecutorBasedEventDrivenDispatcher.scala文件，该文件展示了如何重写createMailbox方法以创建自定义类型的邮箱。 - Viktor Klang

Answer 2

也许您可以使用两个角色来完成这个任务。一个是 facade，另一个是 worker。客户端将作业发送到 facade。然后，facade 将其转发给 worker 并在其内部状态中记住它们，即一个 Set queuedJobs。当它收到一个已经在队列中的作业时，它会将其丢弃。每次 worker 开始处理作业（或者完成作业，根据您的需要），它都会向 facade 发送一个 StartingOn(job) 消息，该消息将从 queuedJobs 中删除该作业。

- Didier Dupont

---

是的，有很多方法可以做到这一点 - 但是看着Java版本非常简单，我期望使用Akka甚至有更简单的方法来做到这一点。也许应该添加的正确问题是：我是否可以通过一个明确定义的API访问邮箱？ - jts

@JTS：我认为你做不到。顺便说一下，didierd的解决方案非常简单，我相信实现的代码不会比你展示的Java解决方案更长。为了简化任务，例如，您可以使用forward，它允许通过保留发送者轻松转发。 - paradigmatic

http://groups.google.com/group/akka-user/browse_thread/thread/67077cad1b2bc4b2/1ff72599f5843e25?lnk=gst&q=mailbox#1ff72599f5843e25 - jts

Answer 3

提议的设计没有意义。最接近Runnable的东西应该是Actor。当然，您可以将它们保存在列表中，如果它们已经存在，则不添加它们。这样的列表由routing actors维护，可以使用Akka提供的准备部件创建，也可以使用forward方法从基本actor创建。

您无法查看另一个actor的邮箱，并且覆盖!没有意义。您要做的是将所有消息发送到路由actor，然后路由actor将其转发到适当的目标。

自然地，由于它接收这些消息，因此可以在那一点上执行任何逻辑。

- Daniel C. Sobral

---

我同意。看起来是最好的选择。我会试一试。 - jts