Samza 事件循环

事件循环在任务之间编排读取和处理消息，检查点，窗口和刷新指标。

默认情况下，Samza 在每个容器中使用单个线程来运行任务。这适合于 CPU 绑定的工作; 要获得更多的 CPU 处理器，只需添加更多的容器。单线程执行还简化了共享任务状态和资源管理。

对于 IO 绑定作业，Samza 支持同步和异步任务的更细微的并行性。对于同步任务（StreamTask 和 WindowableTask），可以通过配置内置线程池 job.container.thread.pool.size 来调度它们并行运行。这适合于阻塞 IO 任务方案。对于异步任务（ AsyncStreamTask ），您可以进行异步 IO 调用，并在完成后触发回调。Samza 提供的最佳并行度在一个任务内，由task.max.concurrency 配置。

最新版本的 Samza 是线程安全的。您可以安全地访问任务线程中的键值存储，写入消息和检查点偏移量的作业状态。如果您在任务之间共享其他数据，例如全局变量或静态数据，则如果可以通过多个线程并发访问数据，例如 StreamTask 在配置的线程池中运行多个线程，则不会线程安全。对于任务中的状态（如成员变量），Samza 保证进程，窗口和提交的相互排他性，因此这些操作之间不会有并发的修改，任何来自一个操作的状态变化都将完全可见。

事件循环内部

容器可能有多个SystemConsumers用于消耗来自不同输入系统的消息。每个 SystemConsumer 在其自己的线程上读取消息，但将消息写入共享进程内消息队列。容器使用此队列将所有消息汇总到事件循环中。

事件循环如下：

1. 从传入消息队列中选择一条消息;
2. 安排适当的任务实例来处理消息;
3. 如果实现 WindowableTask，则任务实例上的 Schedule window（）运行，窗口计时器已被触发;
4. 将 process（）和 window（）调用的任何输出发送到相应的 SystemProducer ;
5. 为 提交间隔 已过的任务写入检查点并刷新状态存储。
6. 如果所有任务实例忙于处理未完成的消息，窗口或检查点，则阻止。

容器在循环中进行，直到它被关闭。

同步任务与异步任务的语义

事件循环的语义在运行同步任务和异步任务时有所不同：

• 对于同步任务（StreamTask 和 WindowableTask），默认情况下，process（）和 window（）将在单个主线程上运行。您可以将 job.container.thread.pool.size 配置为大于 1，并且事件循环将process（）和 window（）在线程池中运行。
• 对于异步任务（AsyncStreamTask），processAsync（）将始终在单个线程中调用，而回调可以从不同的用户线程触发。

在这两种情况下，任务中的默认并发性为1，这意味着每个任务处理中最多只有一个未完成的消息。这保证主题分区中的按顺序消息处理。您可以通过将 task.max.concurrency 配置为大于1来进一步增加它。这允许多个未完成的消息由任务并行处理。此选项会增加任务中的并行性，但可能导致无序处理和完成。

在上述任何一种情况下，以下语义都得到保证（对于事件发生的语义，请参阅此处）：

• 如果 task.max.concurrency = 1，则任务中的每个消息进程完成都将保证发生，在下一次调用同一任务的 process（）/ processAsync（）之前。如果 task.max.concurrency> 1，则不存在这种情况 - 在约束之前，用户应该同步对任务中的任何共享/全局变量的访问。
• 当没有对 process（）/ processAsync（）的调用进行挂起时，将调用 WindowableTask.window（），并且在完成之前不会调度新的 process（）/ processAsync（）调用。因此，保证所有以前的 process（）/ processAsync（）调用在调用 WindowableTask.window（）之前发生。在任何后续 process（）/ processAsync（）调用之前，将保证对WindowableTask.window（）的调用发生。Samza 引擎负责确保及时调用该窗口。
• 检查点保证仅覆盖完全处理的事件。只有当没有待处理的进程（）/ processAsync（）或WindowableTask.window（）调用时才会发生。所有之前的调用发生在检查点和检查点发生之前 - 在所有后续调用之前。

更多详细信息和示例可以在Samza Async API和多线程用户指南中找到。

生命周期

开发人员可以挂接到 SamzaContainer 的生命周期中的唯一方法是通过标准的 InitableTask，ClosableTask，StreamTask / AsyncStreamTask 和 WindowableTask。在需要添加可插入逻辑以包装 StreamTask 的情况下，StreamTask 可以被另一个处理自定义逻辑的 StreamTask 实现包装，然后再调用到包装的 StreamTask 中。

一个具体的例子是一组 StreamTask，它们都希望在其 process（）方法中共享相同的 try / catch 逻辑。可以实现 StreamTask，它包装原始 StreamTasks，并使用适当的 try / catch 逻辑围绕原始 process（）调用。

关注微信下载离线手册