集群分片

    示例项目

    你可以查看「」项目，以了解 Akka 集群分片的实际使用情况。

    简介

    当你需要将 Actor 分布在集群中的多个节点上，并且希望能够使用它们的逻辑标识符与它们进行交互，但不必关心它们在集群中的物理位置时，集群分片（Cluster sharding）非常有用，这也可能随着时间的推移而改变。

    例如，它可以是表示域驱动设计（Domain-Driven Design）术语中聚合根（Aggregate Roots）的 Actor。在这里，我们称这些 Actor 为“实体”。这些 Actor 通常具有持久（durable）状态，但此功能不限于具有持久状态的 Actor。

    集群切分通常在有许多状态 Actor 共同消耗的资源（例如内存）多于一台机器上所能容纳的资源时使用。如果你只有几个有状态的 Actor，那么在集群单例（Cluster Singleton）节点上运行它们可能更容易。

    在这个上下文中，分片意味着具有标识符（称为实体）的 Actor 可以自动分布在集群中的多个节点上。每个实体 Actor 只在一个地方运行，消息可以发送到实体，而不需要发送者知道目标 Actor 的位置。这是通过这个扩展提供的ShardRegion Actor 发送消息来实现的，它知道如何将带有实体 ID 的消息路由到最终目标。

    如果启用了该功能，则集群分片将不会在状态为WeaklyUp的成员上活动。

    • 警告：不要将 Cluster Sharding 与 Automatic Downing 一起使用，因为它允许集群分裂为两个单独的集群，从而导致多个分片和实体启动，每个集群中只有一个节点！详见「」。

    一个示例

    这就是实体 Actor 的样子：

    1. public class Counter extends AbstractPersistentActor {
    3.   public enum CounterOp {
    4.     INCREMENT, DECREMENT
    5.   }
    7.   public static class Get {
    8.     final public long counterId;
    10.     public Get(long counterId) {
    11.       this.counterId = counterId;
    12.     }
    13.   }
    15.   public static class EntityEnvelope {
    16.     final public long id;
    17.     final public Object payload;
    19.     public EntityEnvelope(long id, Object payload) {
    20.       this.id = id;
    21.       this.payload = payload;
    22.     }
    23.   }
    25.   public static class CounterChanged {
    26.     final public int delta;
    28.     public CounterChanged(int delta) {
    29.       this.delta = delta;
    30.     }
    31.   }
    33.   int count = 0;
    35.   // getSelf().path().name() is the entity identifier (utf-8 URL-encoded)
    36.   @Override
    37.   public String persistenceId() {
    38.     return "Counter-" + getSelf().path().name();
    39.   }
    41.   @Override
    42.   public void preStart() throws Exception {
    43.     super.preStart();
    44.     getContext().setReceiveTimeout(Duration.ofSeconds(120));
    45.   }
    47.   void updateState(CounterChanged event) {
    48.     count += event.delta;
    49.   }
    51.   @Override
    52.   public Receive createReceiveRecover() {
    53.     return receiveBuilder()
    54.         .match(CounterChanged.class, this::updateState)
    55.         .build();
    56.   }
    58.   @Override
    59.   public Receive createReceive() {
    60.     return receiveBuilder()
    61.       .match(Get.class, this::receiveGet)
    62.       .matchEquals(CounterOp.INCREMENT, msg -> receiveIncrement())
    63.       .matchEquals(CounterOp.DECREMENT, msg -> receiveDecrement())
    64.       .matchEquals(ReceiveTimeout.getInstance(), msg -> passivate())
    65.       .build();
    66.   }
    68.   private void receiveGet(Get msg) {
    69.     getSender().tell(count, getSelf());
    70.   }
    72.   private void receiveIncrement() {
    73.     persist(new CounterChanged(+1), this::updateState);
    74.   }
    76.   private void receiveDecrement() {
    77.     persist(new CounterChanged(-1), this::updateState);
    78.   }
    80.   private void passivate() {
    81.     getContext().getParent().tell(
    82.         new ShardRegion.Passivate(PoisonPill.getInstance()), getSelf());
    83.   }
    84. }

    上面的 Actor 使用事件源和AbstractPersistentActor中提供的支持来存储其状态。它不必是持久性 Actor，但是如果节点之间的实体发生故障或迁移，那么它必须能够恢复其状态（如果它是有价值的）。

    请注意如何定义persistenceId。Actor 的名称是实体标识符（UTF-8 URL 编码）。你也可以用另一种方式定义它，但它必须是唯一的。

    当使用分片扩展时，你首先要使用ClusterSharding.start方法注册支持的实体类型，通常是在集群中每个节点上的系统启动时。ClusterSharding.start为你提供了可以传递的参考。请注意，如果当前群集节点的角色与在ClusterShardingSettings中指定的角色不匹配，ClusterSharding.start将以代理模式启动ShardRegion。

    1. import akka.japi.Option;
    2. import akka.cluster.sharding.ClusterSharding;
    3. import akka.cluster.sharding.ClusterShardingSettings;
    5. Option<String> roleOption = Option.none();
    6. ClusterShardingSettings settings = ClusterShardingSettings.create(system);
    7. ActorRef startedCounterRegion =
    8.         .start("Counter", Props.create(Counter.class), settings, messageExtractor);

    messageExtractor定义了特定于应用程序的方法，以从传入消息中提取实体标识符和分片标识符。

    1. import akka.cluster.sharding.ShardRegion;
    3. ShardRegion.MessageExtractor messageExtractor =
    4.     new ShardRegion.MessageExtractor() {
    6.       @Override
    7.       public String entityId(Object message) {
    8.         if (message instanceof Counter.EntityEnvelope)
    9.           return String.valueOf(((Counter.EntityEnvelope) message).id);
    10.         else if (message instanceof Counter.Get)
    11.           return String.valueOf(((Counter.Get) message).counterId);
    12.         else return null;
    13.       }
    15.       @Override
    16.       public Object entityMessage(Object message) {
    17.         if (message instanceof Counter.EntityEnvelope)
    18.           return ((Counter.EntityEnvelope) message).payload;
    19.         else return message;
    20.       }
    22.       @Override
    23.       public String shardId(Object message) {
    24.         int numberOfShards = 100;
    25.         if (message instanceof Counter.EntityEnvelope) {
    26.           long id = ((Counter.EntityEnvelope) message).id;
    27.           return String.valueOf(id % numberOfShards);
    28.         } else if (message instanceof Counter.Get) {
    29.           long id = ((Counter.Get) message).counterId;
    30.           return String.valueOf(id % numberOfShards);
    31.         } else {
    32.           return null;
    33.         }
    34.       }
    35.     };

    此示例说明了在消息中定义实体标识符的两种不同方法：

    • Get消息包含标识符本身。
    • EntityEnvelope包含标识符，发送给实体 Actor 的实际消息包装在信封中。

    注意这两种消息类型是如何在上面展示的entityId和entityMessage方法中处理的。发送给实体 Actor 的消息是entityMessage返回的，这使得在需要时可以打开信封（unwrap envelopes）。

    分片是一起管理的一组实体。分组由上文所示的extractShardId函数定义。对于特定的实体标识符，分片标识符必须始终相同。否则，实体 Actor 可能会同时在多个位置意外启动。

    创建一个好的分片算法（sharding algorithm）本身就是一个有趣的挑战。尝试产生一个统一的分布，即在每个分片中有相同数量的实体。根据经验，分片的数量应该比计划的最大集群节点数量大十倍。分片少于节点数量将导致某些节点不会承载任何分片。太多的分片将导致对分片的管理效率降低，例如重新平衡开销，并增加延迟，因为协调器（coordinator）参与每个分片的第一条消息的路由。正在运行的集群中的所有节点上的分片算法必须相同。它可以在停止群集中的所有节点后进行更改。

    一个简单的分片算法在大多数情况下都可以很好地工作，它是以分片的实体标识符模数的hashCode的绝对值为基础的。为了方便起见，ShardRegion.HashCodeMessageExtractor提供了这一功能。

    向实体发送的消息始终通过本地ShardRegion发送。命名实体类型的ShardRegion Actor 引用由ClusterSharding.start返回，也可以使用ClusterSharding.shardRegion检索。如果ShardRegion不知道其位置的话，它将查找实体的分片位置。它将把消息委托给正确的节点，并根据需要创建实体 Actor，即在传递特定实体的第一条消息时。

    它是如何工作的?

    ShardRegion Actor 在集群中的每个节点或标记有特定角色的节点组上启动。ShardRegion由两个特定于应用程序的函数创建，用于从传入消息中提取实体标识符（entity identifier）和分片标识符（shard identifier）。分片是统一管理的一组实体。对于特定分片中的第一条消息，ShardRegion将从中心协调者ShardCoordinator请求分片的位置。

    ShardCoordinator决定哪个ShardRegion将拥有Shard，并通知ShardRegion。区域（region）将确认此请求并将Shard 监督者创建为子 Actor。然后，当Shard Actor 需要时，将创建各个Entities。因此，传入消息通过ShardRegion和Shard传输到目标Entity。

    如果shard home是另一个ShardRegion实例，则消息将转发到该ShardRegion实例。当解析分片的位置时，该分片的传入消息将被缓冲，并在分片所在地（）已知时传递。到已解析分片的后续消息可以立即传递到目标目的地，而不涉及ShardCoordinator。

    一旦知道Shard的位置，ShardRegions就直接发送消息。下面是进入此状态的场景。在场景中，使用以下符号：

    • SC - ShardCoordinator
    • M# - Message 1, 2, 3, 等
    • SR# - ShardRegion 1, 2 3, 等
    • S# - Shard 1 2 3, 等
    • E# - Entity 1 2 3, 等，实体是指由集群分片管理的 Actor。

    场景1：向属于本地 ShardRegion 的未知分片发送消息

    1. 传入消息M1到ShardRegion实例SR1。
    2. M1映射到分片S1。SR1不知道S1，所以它向SC询问S1的位置。
    3. SC回答S1的位置是SR1。
    4. R1为实体E1创建子 Actor，并将S1的缓冲消息发送给E1子 Actor。
    5. 到达R1的S1的所有传入消息都可以由R1处理，而不需要SC。它根据需要创建实体子级，并将消息转发给它们。

    场景2：向属于远程 ShardRegion 的未知分片发送消息

    1. 传入消息M2到ShardRegion实例SR1。
    2. M2映射到S2。SR1不知道S2，所以它向SC询问S2的位置。
    3. SC回答S2的位置是SR2。
    4. SR1将S2的缓冲消息发送到SR2。
    5. 到达SR1的S2的所有传入消息都可以由SR1处理，而不需要SC。它将消息转发到SR2。
    6. SR2接收到S2的消息，询问SC，SC回答S2的位置是SR2，这时我们将回到场景1中（但SR2除外）。

    分片位置

    为了确保特定实体 Actor 的至多一个实例在集群中的某个地方运行，所有节点都具有相同的分片（shard）所在位置视图是很重要的。因此，分片分配决策由中心ShardCoordinator执行，它作为一个集群单例运行，即在所有集群节点中的最老成员上或标记有特定角色的一组节点上执行一个实例。

    决定分片位置的逻辑在可插拔分片分配策略中定义。默认实现ShardCoordinator.LeastShardAllocationStrategy将新的分片分配给ShardRegion，其中以前分配的分片数量最少。此策略可以由特定于应用程序的实现替代。

    分片再平衡

    决定要重新平衡哪些分片的逻辑在可插入分片分配策略（a pluggable shard allocation strategy）中定义。默认实现ShardCoordinator.LeastShardAllocationStrategy从ShardRegion中选择用于切换的分片，其中包含以前分配的大多数碎片。然后，它们将以最少数量的先前分配的分片（即集群中的新成员）分配给ShardRegion。

    对于LeastShardAllocationStrategy，有一个可配置的阈值（rebalance-threshold），说明开始重新平衡时差异必须有多大。在分片最多的区域和分片最少的区域中，分片数量的差异必须大于发生重新平衡的rebalance-threshold。

    当rebalance-threshold为1时，给出了最佳分布，因此通常是最佳选择。更高的阈值意味着更多的分片可以同时重新平衡，而不是一个接一个。这样做的优点是，重新平衡过程可以更快，但缺点是不同节点之间的分片数量（因此负载）可能会显著不同。

    ShardCoordinator中分片位置的状态是持久的，带有「Distributed Data」或「」，可以在故障中幸存。当从集群中删除崩溃或无法访问的协调节点（通过down）时，新的ShardCoordinator单例 Actor 将接管并恢复状态。在这种故障期间，具有已知位置的分片仍然可用，而新（未知）分片的消息将被缓冲，直到新的ShardCoordinator可用。

    消息排序

    只要发送者使用同一个ShardRegion Actor 将消息传递给实体 Actor，消息的顺序就会保持不变。只要没有达到缓冲区限制，消息就会以“最多一次传递”的语义尽最大努力传递，与普通消息发送的方式相同。可靠的端到端（end-to-end）消息传递，通过在「」中使用AtLeastOnceDelivery，可以实现“至少一次传递”的语义。

    开销

    由于到协调器的往返（round-trip），针对新的或以前未使用的分片的消息引入了一些额外的延迟。重新平衡分片也可能增加延迟。在设计特定于应用程序的分片解决方案时，应该考虑这一点，例如，为了避免太细的分片。一旦知道分片的位置，唯一的开销（overhead）就是通过ShardRegion发送消息，而不是直接发送消息。

    协调器的状态和分片「」的状态是持久的，可以在失败中幸存。「Distributed Data」或「」可用于存储。默认情况下使用分布式数据（Distributed Data）。

    使用两种模式时的功能相同。如果你的分片实体本身不使用 Akka 持久化（Persistence），那么使用分布式数据模式更方便，因为你不必为持久性设置和操作单独的数据存储（如 Cassandra）。除此之外，使用一种模式而不使用另一种模式没有主要原因。

    在集群中的所有节点上使用相同的模式很重要，即不可能执行滚动升级来更改此设置。

    此模式通过配置启用（默认情况下启用）：

    1. akka.cluster.sharding.state-store-mode = ddata

    ShardCoordinator的状态将在集群内由分布式数据模块复制，具有WriteMajority/ReadMajority一致性。协调器的状态不持久，它没有存储到磁盘。当集群中的所有节点都已停止时，状态将丢失，也不再需要了。

    记忆实体（Remembering Entities）的状态也是持久的，即存储在磁盘上。存储的实体也会在群集完全重新启动后启动。

    集群分片（Cluster Sharding）使用它自己的每个节点角色的分布式数据Replicator。通过这种方式，可以将所有节点的子集用于某些实体类型，将另一个子集用于其他实体类型。每个这样的复制器（replicator）都有一个包含节点角色的名称，因此集群中所有节点上的角色配置都必须相同，即在执行滚动升级时不能更改角色。

    分布式数据的设置在akka.cluster.sharding.distributed-data部分中配置。对于不同的分片实体类型，不可能有不同的distributed-data设置。

    持久化模式

    此模式通过配置启用：

    1. akka.cluster.sharding.state-store-mode = persistence

    因为它是在集群中运行的，所以必须用分布式日志配置持久化。

    达到最少成员数后启动

    在集群设置akka.cluster.min-nr-of-members或akka.cluster.role.<role-name>.min-nr-of-members时，使用集群分片是很好的。这将推迟分片的分配，直到至少有配置数量的区域已经启动并注册到协调器。这就避免了许多分片被分配到第一个注册的区域，只有在以后才被重新平衡到其他节点。

    有关min-nr-of-members的详细信息，请参阅「How To Startup when Cluster Size Reached」。

    仅代理模式

    ShardRegion Actor 也可以在仅代理模式（proxy only mode）下启动，即它不会承载任何实体本身，但知道如何将消息委托到正确的位置。ShardRegion以仅代理模式使用ClusterSharding.startProxy方法启动。此外，如果当前群集节点的角色与传递给ClusterSharding.start方法的ClusterShardingSettings中指定的角色不匹配时，则ShardRegion将以仅代理模式启动。

    Passivation

    如果实体的状态是持久的，则可以停止不用于减少内存消耗的实体。这是由实体 Actor 的特定于应用程序的实现完成的，例如通过定义接收超时（context.setReceiveTimeout）。如果某个消息在停止时已排队到该实体，则将删除邮箱中排队的消息。为了在不丢失此类消息的情况下支持优雅的钝化（passivation），实体 Actor 可以将ShardRegion.Passivate发送给其父Shard。在Passivate中指定的包装消息将被发送回实体，然后该实体将自行停止。在接收到Passivate和终止实体之间，传入消息将被Shard缓冲。这样的缓冲消息随后被传递到实体的新化身。

    Automatic Passivation

    如果实体使用akka.cluster.sharding.passivate-idle-entity-after设置一段时间没有收到消息，或者通过将ClusterShardingSettings.passivateIdleEntityAfter显式设置为一个合适的时间以保持 Actor 活动，则可以将这些实体配置为自动钝化（automatically passivated）。请注意，只有通过分片发送的消息才会被计算在内，因此直接发送到 Actor 的ActorRef的消息或它发送给自身的消息不会被计算为活动。默认情况下，自动钝化是禁止的。

    Remembering Entities

    通过在调用ClusterSharding.start时将ClusterShardingSettings中的rememberEntities标志设置为true，并确保shardIdExtractor处理Shard.StartEntity(EntityId)，可以使每个Shard中的实体列表持久化，这意味着ShardId必须可以从EntityId中提取。

    1. @Override
    2. public String shardId(Object message) {
    3.   int numberOfShards = 100;
    4.   if (message instanceof Counter.EntityEnvelope) {
    5.     long id = ((Counter.EntityEnvelope) message).id;
    6.     return String.valueOf(id % numberOfShards);
    7.   } else if (message instanceof Counter.Get) {
    8.     long id = ((Counter.Get) message).counterId;
    9.     return String.valueOf(id % numberOfShards);
    10.   } else if (message instanceof ShardRegion.StartEntity) {
    11.     long id = Long.valueOf(((ShardRegion.StartEntity) message).entityId());
    12.     return String.valueOf(id % numberOfShards);
    13.   } else {
    14.     return null;
    15.   }
    16. }

    当配置为记忆实体（remember entities）时，每当Shard重新平衡到另一个节点上或在崩溃后恢复时，它将重新创建以前在该分片中运行的所有实体。要永久停止实体，必须向实体 Actor 的父级发送一条Passivate消息，否则在配置中指定的实体重新启动回退之后，该实体将自动重新启动。

    当rememberEntities设置为false时，Shard不会在重新平衡或从崩溃中恢复后自动重新启动任何实体。只有在Shard中收到实体的第一条消息后，才会启动实体。如果实体停止而不使用Passivate，则不会重新启动。

    请注意，实体本身的状态将不会被恢复，除非它们已被持久化，例如「」。

    当启动/停止实体以及重新平衡分片时，rememberEntities的性能成本相当高。这种成本随着每个分片的实体数量增加而增加，我们目前不建议在每个分片上使用超过 10000 个实体。

    如果需要为实体 Actor 使用其他supervisorStrategy，而不是默认（重新启动）策略，则需要创建一个中间父 Actor，该 Actor 定义子实体 Actor 的supervisorStrategy。

    你以同样的方式启动这样一个监督者（supervisor），就像它是实体 Actor 一样。

    1. ClusterSharding.get(system)
    2.     .start(
    3.         "SupervisedCounter", Props.create(CounterSupervisor.class), settings, messageExtractor);

    请注意，当新消息针对（targeted to）实体时，停止的实体将再次启动。

    优雅地关闭

    你可以将ShardRegion.gracefulShutdownInstance消息发送给ShardRegion Actor，以分发由该ShardRegion承载的所有分片，然后将停止ShardRegion Actor。你可以监控（watch）ShardRegion Actor 以便知道什么时候完成。在此期间，其他区域将以协调器触发重新平衡时的相同方式缓冲这些分片的消息。当分片被停止时，协调器将把这些分片分配到其他地方。

    这是由「」自动执行的，因此是集群成员正常退出进程的一部分。

    删除内部群集分片数据

    集群分片协调器使用 Akka 持久化存储分片的位置。重新启动整个 Akka 集群时，可以安全地删除这些数据。请注意，这不是应用程序数据。

    有一个实用程序akka.cluster.sharding.RemoveInternalClusterShardingData，用于删除此数据。

    • 警告：在运行使用群集分片的 Akka 群集节点时，切勿使用此程序。使用此程序前，请停止所有群集节点。

    如果由于数据损坏而无法启动群集分片协调器，则可能需要删除数据，如果同时意外运行两个群集，例如由于使用自动关闭而存在网络分裂，则可能会发生这种情况。

    • 警告：不要将集群分片（Cluster Sharding）与自动关闭（Automatic Downing）一起使用，因为它允许集群分裂为两个单独的集群，从而导致多个分片和实体启动。

    使用这个程序作为一个独立的 Java 主程序：

    1. java -classpath <jar files, including akka-cluster-sharding>
    2.   akka.cluster.sharding.RemoveInternalClusterShardingData
    3.     -2.3 entityType1 entityType2 entityType3

    该程序包含在 文件中。使用与普通应用程序相同的类路径和配置运行它是最简单的。它可以以类似的方式从 sbt 或 Maven 运行。

    指定实体类型名称（与在ClusterSharding的start方法中使用的名称相同）作为程序参数。

    如果将-2.3指定为第一个程序参数，它还将尝试使用不同的persistenceId删除在Akka 2.3.x中由集群分片（Cluster Sharding）存储的数据。

    配置

    可以使用以下属性配置ClusterSharding扩展。当使用ActorSystem参数创建时，ClusterShardingSettings将读取这些配置属性。还可以修改ClusterShardingSettings或从另一个配置部分创建它，布局如下。ClusterShardingSettings是ClusterSharding扩展的start方法的参数，也就是说，如果需要，每个实体类型都可以配置不同的设置。

    1. # Settings for the ClusterShardingExtension
    2. akka.cluster.sharding {
    4.   # The extension creates a top level actor with this name in top level system scope,
    5.   # e.g. '/system/sharding'
    6.   guardian-name = sharding
    8.   # Specifies that entities runs on cluster nodes with a specific role.
    9.   # If the role is not specified (or empty) all nodes in the cluster are used.
    10.   role = ""
    12.   # When this is set to 'on' the active entity actors will automatically be restarted
    13.   # upon Shard restart. i.e. if the Shard is started on a different ShardRegion
    14.   # due to rebalance or crash.
    15.   remember-entities = off
    17.   # Set this to a time duration to have sharding passivate entities when they have not
    18.   # gotten any message in this long time. Set to 'off' to disable.
    19.   passivate-idle-entity-after = off
    21.   # If the coordinator can't store state changes it will be stopped
    22.   # and started again after this duration, with an exponential back-off
    23.   # of up to 5 times this duration.
    24.   coordinator-failure-backoff = 5 s
    26.   # The ShardRegion retries registration and shard location requests to the
    27.   # ShardCoordinator with this interval if it does not reply.
    28.   retry-interval = 2 s
    30.   # Maximum number of messages that are buffered by a ShardRegion actor.
    31.   buffer-size = 100000
    33.   # Timeout of the shard rebalancing process.
    34.   # Additionally, if an entity doesn't handle the stopMessage
    35.   # after (handoff-timeout - 5.seconds).max(1.second) it will be stopped forcefully
    36.   handoff-timeout = 60 s
    38.   # Time given to a region to acknowledge it's hosting a shard.
    39.   shard-start-timeout = 10 s
    40.   # If the shard is remembering entities and can't store state changes
    41.   # will be stopped and then started again after this duration. Any messages
    42.   # sent to an affected entity may be lost in this process.
    43.   shard-failure-backoff = 10 s
    45.   # If the shard is remembering entities and an entity stops itself without
    46.   # using passivate. The entity will be restarted after this duration or when
    47.   # the next message for it is received, which ever occurs first.
    48.   entity-restart-backoff = 10 s
    50.   # Rebalance check is performed periodically with this interval.
    51.   rebalance-interval = 10 s
    53.   # Absolute path to the journal plugin configuration entity that is to be
    54.   # used for the internal persistence of ClusterSharding. If not defined
    55.   # the default journal plugin is used. Note that this is not related to
    56.   # persistence used by the entity actors.
    57.   # Only used when state-store-mode=persistence
    58.   journal-plugin-id = ""
    60.   # Absolute path to the snapshot plugin configuration entity that is to be
    61.   # used for the internal persistence of ClusterSharding. If not defined
    62.   # the default snapshot plugin is used. Note that this is not related to
    63.   # persistence used by the entity actors.
    64.   # Only used when state-store-mode=persistence
    65.   snapshot-plugin-id = ""
    67.   # Defines how the coordinator stores its state. Same is also used by the
    68.   # shards for rememberEntities.
    69.   # Valid values are "ddata" or "persistence". 
    70.   state-store-mode = "ddata"
    72.   # The shard saves persistent snapshots after this number of persistent
    73.   # events. Snapshots are used to reduce recovery times.
    74.   # Only used when state-store-mode=persistence
    75.   snapshot-after = 1000
    77.   # The shard deletes persistent events (messages and snapshots) after doing snapshot
    78.   # keeping this number of old persistent batches.
    79.   # Batch is of size `snapshot-after`.
    80.   # When set to 0 after snapshot is successfully done all messages with equal or lower sequence number will be deleted.
    81.   # Default value of 2 leaves last maximum 2*`snapshot-after` messages and 3 snapshots (2 old ones + fresh snapshot)
    82.   keep-nr-of-batches = 2
    84.   # Setting for the default shard allocation strategy
    85.   least-shard-allocation-strategy {
    86.     # Threshold of how large the difference between most and least number of
    87.     # allocated shards must be to begin the rebalancing.
    88.     # The difference between number of shards in the region with most shards and
    89.     # the region with least shards must be greater than (>) the `rebalanceThreshold`
    90.     # for the rebalance to occur.
    91.     # 1 gives the best distribution and therefore typically the best choice.
    92.     # Increasing the threshold can result in quicker rebalance but has the
    93.     # drawback of increased difference between number of shards (and therefore load)
    94.     # on different nodes before rebalance will occur.
    95.     rebalance-threshold = 1
    97.     # The number of ongoing rebalancing processes is limited to this number.
    98.     max-simultaneous-rebalance = 3
    99.   }
    101.   # Timeout of waiting the initial distributed state (an initial state will be queried again if the timeout happened)
    102.   # Only used when state-store-mode=ddata
    103.   waiting-for-state-timeout = 5 s
    105.   # Timeout of waiting for update the distributed state (update will be retried if the timeout happened)
    106.   # Only used when state-store-mode=ddata
    107.   updating-state-timeout = 5 s
    109.   # The shard uses this strategy to determines how to recover the underlying entity actors. The strategy is only used
    110.   # by the persistent shard when rebalancing or restarting. The value can either be "all" or "constant". The "all"
    111.   # strategy start all the underlying entity actors at the same time. The constant strategy will start the underlying
    112.   # entity actors at a fix rate. The default strategy "all".
    113.   entity-recovery-strategy = "all"
    115.   # Default settings for the constant rate entity recovery strategy
    116.   entity-recovery-constant-rate-strategy {
    117.     # Sets the frequency at which a batch of entity actors is started.
    118.     frequency = 100 ms
    119.     # Sets the number of entity actors to be restart at a particular interval
    120.     number-of-entities = 5
    121.   }
    123.   # Settings for the coordinator singleton. Same layout as akka.cluster.singleton.
    124.   # The "role" of the singleton configuration is not used. The singleton role will
    125.   # be the same as "akka.cluster.sharding.role".
    126.   coordinator-singleton = ${akka.cluster.singleton}
    128.   # Settings for the Distributed Data replicator. 
    129.   # Same layout as akka.cluster.distributed-data.
    130.   # The "role" of the distributed-data configuration is not used. The distributed-data
    131.   # role will be the same as "akka.cluster.sharding.role".
    132.   # Note that there is one Replicator per role and it's not possible
    133.   # to have different distributed-data settings for different sharding entity types.
    134.   # Only used when state-store-mode=ddata
    135.   distributed-data = ${akka.cluster.distributed-data}
    136.   distributed-data {
    137.     # minCap parameter to MajorityWrite and MajorityRead consistency level.
    138.     majority-min-cap = 5
    139.     durable.keys = ["shard-*"]
    141.     # When using many entities with "remember entities" the Gossip message
    142.     # can become to large if including to many in same message. Limit to
    143.     # the same number as the number of ORSet per shard.
    144.     max-delta-elements = 5
    146.   }
    148.   # The id of the dispatcher to use for ClusterSharding actors.
    149.   # If not specified default dispatcher is used.
    150.   # If specified you need to define the settings of the actual dispatcher.
    151.   # This dispatcher for the entity actors is defined by the user provided
    152.   # Props, i.e. this dispatcher is not used for the entity actors.
    153.   use-dispatcher = ""
    154. }

    自定义分片分配策略（shard allocation strategy）可以在ClusterSharding.start的可选参数中定义。有关如何实现自定义分片分配策略的详细信息，请参阅AbstractShardAllocationStrategy的 API 文档。

    检查群集分片状态

    有两个检查群集状态的请求可用：

    • ShardRegion.getShardRegionStateInstance，它将返回一个ShardRegion.ShardRegionState，其中包含区域中运行的分片的标识符以及每个分片的活动实体。
    • ShardRegion.GetClusterShardingStats，它将查询集群中的所有区域，并返回一个ShardRegion.ClusterShardingStats，其中包含每个区域中运行的分片的标识符以及每个分片中活动的实体数。

    可以通过ClusterSharding.getShardTypeNames获取所有已启动分片的类型名。

    这些消息的目的是测试（testing）和监控（monitoring），它们不提供直接向各个实体发送消息的访问权。

    在进行滚动升级（rolling upgrades）时，必须特别注意不要改变以下任何分片方面：

    • 函数
    • 分片区域运行的角色

    如果其中任何一个需要更改，则需要完全重新启动群集。

    英文原文链接：.