从Java并发迁移到Scala并发

Question

我在Java中有一个相当标准的机制来解决这个问题：

• 工作项必须安排在特定时间执行。
• 然后，每个工作项都必须在条件变为真时等待。
• 工作项应该是可取消的

我使用的解决方案如下：

1. 有一个单线程调度程序来调度我的工作项。
2. 有一个ExecutorService (可能是多线程的)
3. 然后，每个计划工作项向ExecutorService提交实际工作。返回的Future被缓存在映射中。完成服务用于在工作完成后从缓存中删除未来。
4. 项目可通过缓存的期货取消。

当然，我的执行者至少需要和我希望拥有的阻塞工作项的数量一样多，但这在实践中并不是一个问题。

现在，我正在Scala中编写代码，并使用actor框架。假设我的工作项可以封装在发送给参与者的事件中：

1. 我将使用什么机制来安排特定时间的工作项？
2. 如果工作项是发送给参与者的事件，如何确保备份线程池大于可以同时阻塞的项数。
3. 如何使先前计划的工作项被取消？

Answer 1

我将使用什么机制来安排特定时间的工作项？

我会用java.util.concurrent.ScheduledExecutorService.

如果工作项是发送给参与者的事件，如何确保备份线程池大于可以同时阻塞的项数。

在我看来，这是一种挫败并行化努力的设计。尽量减少或消除阻塞和全局状态。这些都是可组合性和可伸缩性的障碍。例如，考虑有一个单独的专用线程，该线程等待文件到达，然后将事件发送给参与者。或者查看用于异步非阻塞I/O的java.nio。

我在这里并不完全理解您的需求，但是您似乎可以有一个线程/参与者来查找I/O事件。然后，作为您预定的“工作项”，计划创建非阻塞角色的效果。让这些参与者向I/O线程/参与者注册，以接收有关他们所关心的I/O事件的消息。

如何使先前计划的工作项被取消？

ScheduledExecutorService返回期货。在这方面，你拥有的不是一个糟糕的设计。在地图中收集它们并调用future.cancel()。

Answer 2

您可以有一个调度参与者，它有一个调度的参与者列表，并使用Actor.receiveWithin()每一秒左右唤醒一次，并向准备执行的参与者发送消息。调度器也可以处理取消。另一种选择是让每个参与者使用receiveWithin()直接处理自己的调度，而不是集中调度。

在博客文章Scala中简单的类cron调度程序中有一些关于这个问题的讨论。