协作

  • 警告：此模块当前标记为「可能更改」。它已准备好用于生产，但 API 可能在没有警告或预测期的情况下发生变化。

Akka 协作（Coordination）是一套用于分布式协作的工具。

Discovery曾是 Akka 管理的一部分，但从 Akka 的2.5.19版和 Akka 管理的1.0.0版开始，它已成为 Akka 的模块。如果你还将 Akka 管理用于其他服务发现方法或加载程序，请确保你至少使用了 Akka 管理（Management）的1.0.0版本。

依赖

为了使用 Akka 协作，你需要将以下依赖添加到你的项目中：

  1. <!-- Maven -->
  2. <dependency>
  3.   <groupId>com.typesafe.akka</groupId>
  4.   <artifactId>akka-coordination_2.12</artifactId>
  5.   <version>2.5.23</version>
  6. </dependency>
  8. <!-- Gradle -->
  9. dependencies {
  10.   compile group: 'com.typesafe.akka', name: 'akka-coordination_2.12', version: '2.5.23'
  11. }
  13. <!-- sbt -->
  14. libraryDependencies += "com.typesafe.akka" %% "akka-coordination" % "2.5.23"

Lease

Lease是分布式锁的可插拔 API。

使用 Lease

Lease通过以下方式加载：

  • Lease名称
  • 配置位置以指示应加载哪个实现
  • 所有者名称

任何Lease实现应提供以下保证：

  • 同一名称的Lease多次加载，即使在不同的节点上，也是同一Lease
  • 一次只能有一个所有者获得Lease

获取Lease：

  1. Lease lease =
  2.     LeaseProvider.get(system).getLease("<name of the lease>", "docs-lease", "<owner name>");
  3. CompletionStage<Boolean> acquired = lease.acquire();
  4. boolean stillAcquired = lease.checkLease();
  5. CompletionStage<Boolean> released = lease.release();

获取Lease会返回一个CompletionStage，因为Lease实现通常是通过第三方系统（如 Kubernetes API 服务器或 ZooKeeper）实现的。

一旦获得Lease，就可以调用checkLease以确保Lease仍然被获得。由于Lease实现基于其他分布式系统，因此Lease可能会因第三方系统超时而丢失。此操作不是异步的，因此可以在执行任何具有Lease的操作之前调用它。

Lease有一个所有者。如果同一所有者多次尝试获取Lease，那么它将成功，即Lease是可重入的。

选择一个对你的用例来说是唯一的Lease名称是很重要的。如果集群中每个节点的Lease都需要唯一，那么可以使用群集主机端口：

  1. String owner = Cluster.get(system).selfAddress().hostPort();

对于在同一节点上有多个不同Lease的用例，必须在名称中添加一些唯一的东西。例如，Lease可以与集群分片（Cluster Sharding）一起使用，在这种情况下，分片 ID 包含在每个分片的Lease名称中。

其他 Akka 模块的用途

Lease可以用于「集群单例」和「集群分片」。

Lease 实现

实现一个 Lease

实现应该扩展akka.coordination.lease.javadsl.Lease

  1. static class SampleLease extends Lease {
  3.   private LeaseSettings settings;
  5.   public SampleLease(LeaseSettings settings) {
  6.     this.settings = settings;
  7.   }
  9.   @Override
  10.   public LeaseSettings getSettings() {
  11.     return settings;
  12.   }
  14.   @Override
  15.   public CompletionStage<Boolean> acquire() {
  16.     return null;
  17.   }
  19.   @Override
  20.   public CompletionStage<Boolean> acquire(Consumer<Optional<Throwable>> leaseLostCallback) {
  21.     return null;
  22.   }
  24.   @Override
  25.   public CompletionStage<Boolean> release() {
  26.     return null;
  27.   }
  29.   @Override
  30.   public boolean checkLease() {
  31.     return false;
  32.   }
  33. }

这些方法应提供以下保证：

  • acquire：如果成功获取Lease，则为true；如果Lease被其他所有者占用，则为false；如果无法与实现Lease的第三方系统通信，则失败。
  • release：如果Lease已明确释放，则为true；如果Lease未明确释放，则为false；如果不知道Lease是否已释放，则失败。
  • 在acquire的CompletionStage完成之前，checkLease应返回false；如果Lease由于与第三方通信错误而丢失，也应该返回false。检查Lease也不应阻塞。
  • Lease lost callback只能在acquire的CompletionStage完成后调用，如果Lease丢失（例如，由于与第三方系统的通信丢失），则应调用该回调。

此外，期望Lease的实现包括生存时间机制（a time to live mechanism），这意味着在节点崩溃的情况下不会永远保留Lease。如果用户希望在这种情况下进行外部干预以获得最大的安全性，那么生存时间可以设置为无限。

配置必须为Lease实现的 FQCN 定义lease-class属性。

如果默认值来自akka.coordination.lease，则Lease实现应支持以下属性：

  1. # if the node that acquired the leases crashes, how long should the lease be held before another owner can get it
  2. heartbeat-timeout = 120s
  4. # interval for communicating with the third party to confirm the lease is still held
  5. heartbeat-interval = 12s
  7. # lease implementations are expected to time out acquire and release calls or document
  8. # that they do not implement an operation timeout
  9. lease-operation-timeout = 5s

此配置位置将传递到getLease。

  1. akka.actor.provider = cluster
  2. docs-lease {
  3.   lease-class = "docs.akka.coordination.SampleLease"
  4.   heartbeat-timeout = 100s
  5.   heartbeat-interval = 1s
  6.   lease-operation-timeout = 1s
  7.   # Any lease specific configuration
  8. }

英文原文链接：Coordination.