集群

Camel 提供了以下与集群相关的 SPI：

• 集群服务（Cluster Service）

  一个常规的 Camel 服务，用于管理集群资源，例如视图（见下文）

• 集群视图（Cluster View）

  表示集群的一个视图，拥有自己独立的一组资源。目前视图提供以下支持：

  • 领导者选举
  • 拓扑事件，例如成员加入/离开集群

• 集群成员（Cluster Member）

  表示集群中的一个成员。

集群 SPI 配置

集群服务与其他 Camel 服务并无二致，因此要对其进行配置，只需将你的实现注册到 CamelContext 中：

仅 Java：在 CamelContext 上注册集群服务

MyClusterServiceImpl service = new MyClusterServiceImpl();
context.addService(service);

Cluster Service（集群服务）的配置取决于你所选择的实现。Camel 开箱即提供了以下几种实现：

配置选项：

ConsulClusterService

FileLockClusterService

将集群根目录存储在 NFS 上

当将 FileLockClusterService 的集群根目录（参见 rootPath 配置选项）存储在 NFS 上时，所有集群成员必须使用允许故障切换可靠工作的 NFS 客户端挂载选项。

以下 NFS 客户端挂载选项有助于实现这一点。你可以根据实际需要调整这些选项。

• soft — 当 NFS 服务器未响应请求时，禁止客户端持续进行重传尝试。
• timeo=10 — NFS 客户端在再次发送请求之前，等待 NFS 服务器响应的时间，单位为十分之一秒。默认值为 600（60 秒）。
• retrans=1 — 指定 NFS 客户端尝试向 NFS 服务器重传失败请求的次数。默认值为 3。客户端在每次 retrans 尝试之间会等待一个 timeo 超时周期。
• lookupcache=none — 指定内核如何管理挂载点的目录项缓存。none 会强制客户端在使用所有缓存项之前对其进行重新验证。这使集群领导者能够立即检测到对其锁文件所做的任何更改，并防止锁检查机制返回不正确的有效性信息。
• sync — 任何向挂载点上的文件写入数据的系统调用，都会在将控制权返回给用户空间之前先将数据刷新到 NFS 服务器。该选项可提供更高的数据缓存一致性。

有关 NFS 挂载选项的更多信息，请参阅 http://linux.die.net/man/5/nfs。

InfinispanClusterService

JGroupsRaftClusterService

KubernetesClusterService

ZooKeeperClusterService

配置示例：

• Spring Boot

  camel.cluster.file.enabled = true
  camel.cluster.file.id = ${random.uuid}
  camel.cluster.file.root = ${java.io.tmpdir}

• Spring XML

  <?xml version="1.0" encoding="UTF-8"?>
  <beans xmlns="http://www.springframework.org/schema/beans"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="
           http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans.xsd
           http://camel.apache.org/schema/spring
           http://camel.apache.org/schema/spring/camel-spring.xsd">
  
    <bean id="zx" class="org.apache.camel.component.zookeeper.cluster.ZooKeeperClusterService">
      <property name="id" value="node-1"/>
      <property name="basePath" value="/camel/cluster"/>
      <property name="nodes" value="localhost:2181"/>
    </bean>
  
    <camelContext xmlns="http://camel.apache.org/schema/spring" autoStartup="false">
      ...
    </camelContext>
  
  </beans>

Cluster SPI 的使用

Cluster SPI 被以下新增实现所使用：

• ClusteredRoutePolicy

  这是 RoutePolicy 的一个实现，当它所使用的集群视图（Cluster View）获得领导权时，会启动与其关联的路由。

  仅 Java：在路由中使用 ClusteredRoutePolicy

  context.addRoutes(new RouteBuilder {
      @Override
      public void configure() throws Exception {
          // Create the route policy
          RoutePolicy policy = ClusteredRoutePolicy.forNamespace("my-ns");
  
          // bind the policy to one or more routes
          from("timer:clustered?delay=1000&period=1000")
              .routePolicy(policy)
              .log("Route ${routeId} is running ...");
      }
  });

要将同一策略应用于所有路由，可以使用专用的 RoutePolicyFactory。

仅 Java：使用 ClusteredRoutePolicyFactory

// 将集群路由策略工厂添加到上下文
context.addRoutePolicyFactory(ClusteredRoutePolicyFactory.forNamespace("my-ns"));

context.addRoutes(new RouteBuilder {
    @Override
    public void configure() throws Exception {
        // 将策略绑定到一个或多个路由
        from("timer:clustered?delay=1000&period=1000")
            .log("Route ${routeId} is running ...");
    }
});

• ClusteredRouteController

  这是 RouteController SPI 的一种实现，它让 camel context 先启动，然后在取得／失去领导权时启动／停止路由。该实现与 spring-boot 应用集成良好，假设你的路由配置如下：

  仅 Java：ClusteredRouteController 的 Spring Boot 路由定义

  @Bean
  public RouteBuilder routeBuilder() {
      return new RouteBuilder() {
          @Override
          public void configure() throws Exception {
              from("timer:heartbeat?period=10000")
                  .routeId("heartbeat")
                  .log("HeartBeat route (timer) ...");
              from("timer:clustered?period=5000")
                  .routeId("clustered")
                  .log("Clustered route (timer) ...");
          }
      };
  }

然后，你可以利用 Spring Boot 配置让它们组成集群：

# 启用路由控制器
camel.clustered.controller.enabled = true

# 定义路由的默认命名空间
camel.clustered.controller.namespace = my-ns

# 将 id 为 'heartbeat' 的路由排除在集群之外
camel.clustered.controller.routes[heartbeat].clustered = false

• Master 组件

  master 组件与 ClusteredRoutePolicy 类似，但它工作在消费者级别，因此能确保在任何时刻集群中只有一个端点在消费资源。它的配置非常简单，你只需按照 master 组件的语法为单例端点加上前缀即可：

  master:namespace:delegateUri

具体示例：

仅限 Java：使用 master 组件的 Spring Boot 路由

@Bean
public RouteBuilder routeBuilder() {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("timer:heartbeat?period=10000")
                .routeId("heartbeat")
                .log("HeartBeat route (timer) ...");

            from("master:my-ns:timer:clustered?period=5000")
                .routeId("clustered")
                .log("Clustered route (timer) ...");
        }
    };
}