主页
产品
Red Hat Fuse
7.12
Apache Camel 组件参考
191.7. 使用 Kafka idempotent 存储库

191.7. 使用 Kafka idempotent 存储库

Camel 2.19 中可用

camel-kafka 库提供了一个基于 Kafka 主题的幂等存储库。此存储库将所有更改广播到 Kafka 主题中的幂等状态(add/remove)，并通过事件源为每个存储库的进程实例填充本地内存缓存。

使用的主题必须为每个幂等存储库实例唯一。机制对于主题分区的数量没有任何要求；因为存储库同时消耗所有分区。它还对主题的复制因子没有任何要求。

每个使用主题的存储库实例（例如，通常在并行运行的不同机器上）控制自己的消费者组，因此，使用相同主题的 10 Camel 进程集群将控制其自身偏移。

启动时，实例会订阅主题，并将偏移重新放在开始，将缓存重新构建到最新的状态。缓存不会被视为温，直到长度为 pollDurationMs 轮询返回 0 记录。在缓存已温启动或 30 秒前，启动才会完成；如果后者发生幂等存储库可能处于不一致的状态，直到其消费者捕获到主题的末尾为止。

KafkaIdempotentRepository 具有以下属性：

Expand

属性	描述
`topic`	用于广播更改的 Kafka 主题的名称。（必需）
`bootstrapServers`	内部 Kafka producer 和 consumer 上的 `bootstrap.servers` 属性。如果没有设置 `consumerConfig` 和 `producerConfig`，则将其用作简短的。如果使用，此组件将为生成者和消费者应用可诊断的默认配置。
`producerConfig`	设置用于广播更改的 Kafka producer 使用的属性。覆盖 `bootstrapServers`，因此必须定义 Kafka `bootstrap.servers` 属性本身
`consumerConfig`	设置 Kafka 消费者用于从主题中填充缓存的属性。覆盖 `bootstrapServers`，因此必须定义 Kafka `bootstrap.servers` 属性本身
`maxCacheSize`	最近使用的密钥应存储在内存中（默认值 1000）。
`pollDurationMs`	Kafka 消费者的轮询持续时间。本地缓存会立即更新。这个值会影响从主题更新其缓存的其他同级服务器相对于发送缓存的幂等消费者实例。默认值为 100 ms。如果明确设置这个值，请注意远程缓存存活度和此存储库消费者和 Kafka 代理之间的网络流量卷之间有一个利弊。缓存温过程还取决于有一个轮询，没有任何内容 - 这表示流已被消耗到当前点。如果轮询持续时间过长，对于在主题上发送消息的速率，则缓存可能会进行温处理，并且与其对等点相比将以不一致的状态运行，直到捕获为止。

可以通过定义 topic 和 bootstrapServers，或者 producerConfig 和 consumerConfig 属性集来实例化存储库，以启用 SSL/SASL 等功能。

要使用，此存储库必须手动或注册为 Spring/Blueprint 中的 bean，因为它是 CamelContext aware。

示例用法如下：

KafkaIdempotentRepository kafkaIdempotentRepository = new KafkaIdempotentRepository("idempotent-db-inserts", "localhost:9091");

SimpleRegistry registry = new SimpleRegistry();
registry.put("insertDbIdemRepo", kafkaIdempotentRepository); // must be registered in the registry, to enable access to the CamelContext
CamelContext context = new CamelContext(registry);

// later in RouteBuilder...
from("direct:performInsert")
    .idempotentConsumer(header("id")).messageIdRepositoryRef("insertDbIdemRepo")
        // once-only insert into database
    .end()

KafkaIdempotentRepository kafkaIdempotentRepository = new KafkaIdempotentRepository("idempotent-db-inserts", "localhost:9091");

SimpleRegistry registry = new SimpleRegistry();
registry.put("insertDbIdemRepo", kafkaIdempotentRepository); // must be registered in the registry, to enable access to the CamelContext
CamelContext context = new CamelContext(registry);

// later in RouteBuilder...
from("direct:performInsert")
    .idempotentConsumer(header("id")).messageIdRepositoryRef("insertDbIdemRepo")
        // once-only insert into database
    .end()

Copy to Clipboard

Toggle word wrap

在 XML 中：

<!-- simple -->
<bean id="insertDbIdemRepo"
  class="org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository">
  <property name="topic" value="idempotent-db-inserts"/>
  <property name="bootstrapServers" value="localhost:9091"/>
</bean>

<!-- complex -->
<bean id="insertDbIdemRepo"
  class="org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository">
  <property name="topic" value="idempotent-db-inserts"/>
  <property name="maxCacheSize" value="10000"/>
  <property name="consumerConfig">
    <props>
      <prop key="bootstrap.servers">localhost:9091</prop>
    </props>
  </property>
  <property name="producerConfig">
    <props>
      <prop key="bootstrap.servers">localhost:9091</prop>
    </props>
  </property>
</bean>

<!-- simple -->
<bean id="insertDbIdemRepo"
  class="org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository">
  <property name="topic" value="idempotent-db-inserts"/>
  <property name="bootstrapServers" value="localhost:9091"/>
</bean>

<!-- complex -->
<bean id="insertDbIdemRepo"
  class="org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository">
  <property name="topic" value="idempotent-db-inserts"/>
  <property name="maxCacheSize" value="10000"/>
  <property name="consumerConfig">
    <props>
      <prop key="bootstrap.servers">localhost:9091</prop>
    </props>
  </property>
  <property name="producerConfig">
    <props>
      <prop key="bootstrap.servers">localhost:9091</prop>
    </props>
  </property>
</bean>

Copy to Clipboard

Toggle word wrap

返回顶部

191.7. 使用 Kafka idempotent 存储库

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links