主页
产品
Red Hat Fuse
7.13
Apache Camel 组件参考
191.7. 使用 Kafka idempotent 软件仓库

191.7. 使用 Kafka idempotent 软件仓库

Camel 2.19 中可用

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

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

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

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

KafkaIdempotentRepository 具有以下属性：

Expand

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

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

要使用，此存储库必须放在 Camel registry 中，也可以手动或通过注册作为 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