29.3. 组件选项

additionalProperties (common)

如果无法直接在 camel 配置上设置 kafka consumer 或 kafka producer 的额外属性（例如：尚未反映在 Camel 配置中的新 Kafka 属性），属性必须加上额外的Properties。e.g: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro。

map

代理 (common)

要使用的 Kafka 代理的 URL。格式为 host1:port1,host2:port2，列表可以是代理的子集，也可以是指向代理子集的 VIP。这个选项在 Kafka 文档中称为 bootstrap.servers。

字符串

clientID (common)

客户端 id 是每个请求中发送的用户指定的字符串，以帮助追踪调用。它应该以逻辑方式识别提出请求的应用。

字符串

配置 (common)

允许使用端点将重复使用的通用选项预配置 Kafka 组件。

KafkaConfiguration

headerFilterStrategy (common)

使用自定义 HeaderFilterStrategy 过滤标题到 Camel 消息或从 Camel 消息过滤。

HeaderFilterStrategy

reconnectBackoffMaxMs (common)

当重新连接到重复连接失败的代理时，需要等待的最大时间（毫秒）。如果提供，每个主机的 backoff 将为每个连续的连接失败指数增加，直到最高值。在计算 backoff 增长后，会添加 20% 随机的 jitter 以避免连接 storms。

1000

整数

shutdownTimeout (common)

超时时间（毫秒），以毫秒为单位，等待消费者或制作者关闭并终止其 worker 线程。

30000

int

allowManualCommit (consumer)

是否允许通过 KafkaManualCommit 进行手动提交。如果启用了这个选项，则 KafkaManualCommit 实例存储在 Exchange 消息标头中，它允许最终用户访问此 API 并通过 Kafka 使用者执行手动偏移提交。

false

布尔值

autoCommitEnable (consumer)

如果为 true，请定期提交到 ZooKeeper，以偏移已由消费者获取的信息。当进程失败时，会使用这个提交的偏移，因为新消费者开始的位置。

true

布尔值

autoCommitIntervalMs (consumer)

消费者偏移的 ms 频率被提交到 zookeeper。

5000

整数

autoCommitOnStop (consumer)

当消费者停止时是否执行显式自动提交，以确保代理具有上次消耗的消息的提交。这要求打开自动CommitEnable 选项。可能的值有： sync、async 或 none。和 sync 是默认值。

Enum 值：

sync

字符串

autoOffsetReset (consumer)

当 ZooKeeper 中没有初始偏移时，或者偏移范围不足： earliest : 自动将偏移重置为最早偏移最新的 : 自动将偏移重置为最新的偏移失败：将偏移重置为最新的偏移失败：向消费者抛出异常。

Enum 值：

latest

字符串

breakOnFirstError (consumer)

此选项控制消费者处理交换时会发生什么，并且失败。如果选项为 false，则使用者将继续执行下一个消息并处理它。如果选项为 true，则消费者会中断，并将返回到导致失败的消息偏移，然后重新尝试处理此消息。但是，如果其绑定每次都失败，这可能会导致无法处理同一消息，例如中毒消息。因此，建议您使用 Camel 的错误处理程序处理示例。

false

布尔值

bridgeErrorHandler (consumer)

允许将消费者桥接到 Camel 路由错误处理程序，这意味着当消费者试图选择传入消息或类似信息时发生异常，现在将作为消息处理并由路由 Error Handler 处理。默认情况下，使用者将使用 org.apache.camel.spi.ExceptionHandler 来处理例外情况，该处理程序将被记录在 WARN 或 ERROR 级别，并忽略。

false

布尔值

checkCrcs (consumer)

自动检查已使用的记录的 CRC32。这样可确保发生消息的在线或磁盘损坏。此检查添加了一些开销，因此在寻求极端性能的情况下可能会禁用它。

true

布尔值

commitTimeoutMs (consumer)

代码将等待同步提交完成的最长时间（以毫秒为单位）。

5000

Long

consumerRequestTimeoutMs (consumer)

配置控制客户端等待请求响应的最长时间。如果响应在超时前收到，客户端会在必要时重新发送请求，或者在重试已耗尽时失败请求。

40000

整数

consumer Count （消费者）

连接到 kafka 服务器的消费者数量。每个使用者在单独的线程上运行，后者检索和处理传入的数据。

1

int

fetchMaxBytes (consumer)

服务器应该为获取请求返回的最大数据量。如果 fetch 的第一个非空分区中的第一个消息大于这个值，则消息仍会返回，以确保消费者进行进度。代理接受的最大消息大小通过 message.max.bytes (broker config)或 max.message.bytes (topic config)定义。请注意，消费者并行执行多个获取。

52428800

整数

fetchMinBytes (consumer)

服务器应该为获取请求返回的最小数据量。如果数据不足，请求将在回答请求之前等待大量数据。

1

整数

fetchWaitMaxMs (consumer)

如果没有足够的数据来满足 fetch.min.bytes，则服务器在回答获取请求前将阻止的最大时间。

500

整数

groupId (consumer)

唯一标识此使用者进程组的字符串。通过设置相同的组 id 多个进程，表示它们都是同一消费者组的一部分。消费者需要这个选项。

字符串

groupInstanceId (consumer)

最终用户提供的使用者实例的唯一标识符。仅允许非空字符串。如果设置，使用者将被视为静态成员，这意味着任何消费者组中只允许具有此 ID 的一个实例。这可以与更大的会话超时结合使用，以避免由临时不可用（如进程重启）导致的组重新平衡。如果没有设置，使用者将作为动态成员加入组，这是传统的行为。

字符串

headerDeserializer (consumer)

使用自定义 KafkaHeaderDeserializer 来反序列化 kafka 标头值。

KafkaHeaderDeserializer

heartbeatIntervalMs (consumer)

使用 Kafka 的组管理功能时，心跳到消费者协调器之间的预期时间。心跳用于确保消费者的会话保持活跃状态，并在新消费者加入或离开组时进行重新平衡。该值必须小于 session.timeout.ms，但设置的值通常不应超过该值的 1/3。它甚至可以降低以控制正常重新平衡的预期时间。

3000

整数

keyDeserializer (consumer)

为实现 Deserializer 接口的密钥反序列化器类。

org.apache.kafka.common.serialization.StringDeserializer

字符串

maxPartitionFetchBytes (consumer)

服务器将返回的每个分区的最大数据量。请求使用的最大内存为 #partitions max.partition.fetch.bytes。此大小必须至少与服务器允许的最大消息大小相同，否则生成者可能会发送大于消费者的消息。如果发生这种情况，使用者可能会卡住尝试在特定分区上获取大型消息。

1048576

整数

maxPollIntervalMs （使用者）

使用消费者组管理时轮询（）调用之间的最大延迟。这会在获取更多记录前，在消费者闲置的时间上绑定上。如果在此超时过期之前没有调用 poll （），那么消费者将被视为失败，并且组将重新平衡，以便将分区重新分配给另一个成员。

Long

maxPollRecords (consumer)

在要 poll （）的单个调用中返回的最大记录数。

500

整数

offsetRepository (consumer)

要使用的偏移存储库，以便本地存储主题的每个分区的偏移量。定义一个将禁用 autocommit。

StateRepository

partitionAssignor (consumer)

使用组管理时，客户端将使用的分区分配策略的类名称在消费者实例之间分发分区所有权。

org.apache.kafka.clients.consumer.RangeAssignor

字符串

pollOnError (consumer)

如果 kafka threw 异常在轮询新消息时，应该做什么。默认情况下，将使用组件配置中的值，除非在端点级别上配置了显式值。DISCARD 将丢弃消息并继续轮询下一个消息。ERROR_HANDLER 将使用 Camel 的错误处理程序来处理异常，之后继续轮询下一个消息。RECONNECT 将重新连接消费者，并尝试再次轮询消息，使消费者再次轮询同一消息将停止消费者（如果消费者应可以再次使用，则必须手动启动/重新启动）。

Enum 值：

ERROR_HANDLER

PollOnError

pollTimeoutMs (consumer)

轮询 KafkaConsumer 时使用的超时。

5000

Long

resumeStrategy (consumer)

这个选项允许用户设置自定义恢复策略。当分配分区时（例如：连接或重新连接时），执行恢复策略。它允许实施自定义如何恢复操作，并作为 seekTo 和 offsetRepository 机制的更灵活的替代方案。有关实现详情，请参阅 KafkaConsumerResumeStrategy。此选项不会影响自动提交设置。使用此设置的实现可能还要使用手动提交选项来评估。

KafkaConsumerResumeStrategy

seekTo (consumer)

设置 KafkaConsumer 将在启动时从开始或结尾读取：从开始开始读取 : read from end : read from end This is replace the earlier attribute seekToBeginning。

Enum 值：

字符串

sessionTimeoutMs (consumer)

使用 Kafka 组管理功能时用于检测故障的超时。

10000

整数

specificAvroReader (consumer)

这可让特定的 Avro reader 用于 Confluent Platform schema registry 和 io.confluent.kafka.serializers.KafkaAvroDeserializer。这个选项只在 Confluent Platform （不是标准 Apache Kafka）中可用。

false

布尔值

topicIsPattern (consumer)

主题是否为模式（正则表达式）。这可以用于订阅与模式匹配的动态主题数。

false

布尔值

值Deserializer （使用者）

deserializer 类用于实现 Deserializer 接口的值。

org.apache.kafka.common.serialization.StringDeserializer

字符串

kafkaManualCommitFactory （使用者（高级）

Autowired Factory 用于创建 KafkaManualCommit 实例。当从开箱即用的默认实现中手动提交时，这允许自定义工厂创建自定义 KafkaManualCommit 实例，以防需要进行特殊逻辑。

KafkaManualCommitFactory

pollExceptionStrategy (consumer (advanced))

Autowired To 使用带有消费者的自定义策略来控制如何在池消息时处理 Kafka 代理中的异常。

PollExceptionStrategy

bufferMemorySize (producer)

制作者可用于缓冲记录的总字节，用于等待发送到服务器。如果记录发送速度快于服务器，则生产者将根据 block.on.buffer.full 指定的首选项来阻止或抛出异常。此设置应当与制作者要使用的内存大致对应，但并非者所使用的所有内存都不会被硬绑定。一些额外的内存将用于压缩（如果启用了压缩），以及维护 in-flight 请求。

33554432

整数

compressionCodec (producer)

此参数允许您为此制作者生成的所有数据指定压缩代码。有效值为 none、gzip 和 snappy。

Enum 值：

none

字符串

connectionMaxIdleMs (producer)

在此配置指定的毫秒数后关闭闲置连接。

540000

整数

deliveryTimeoutMs (producer)

在调用 send （）后报告成功或失败的时间上限。这限制了在发送前延迟记录的总时间，从代理等待确认时间（如预期），以及可重新发送失败的时间。

120000

整数

enableIdempotence (producer)

如果设置为 'true'，则制作者将确保在流中写入每个消息的确切副本。如果为 'false'，则制作者重试可能会在流中写入重试消息的重复。如果设置为 true，这个选项将需要 max.in.flight.requests.per.connection 设置为 1，重试不能为零，另外的 acks 必须设为 'all'。

false

布尔值

headerSerializer (producer)

使用自定义 KafkaHeaderSerializer 来序列化 kafka 标头值。

KafkaHeaderSerializer

Key (producer)

记录密钥（如果没有指定任何密钥，则为 null）。如果配置了这个选项，则其优先于标头 KafkaConstants#KEY。

字符串

keySerializer (producer)

键的序列化类（如果没有提供，默认为与消息相同）。

org.apache.kafka.common.serialization.StringSerializer

字符串

lazyStartProducer (producer)

制作者是否应启动 lazy （在第一个消息中）。通过启动 lazy，您可以使用此选项来允许 CamelContext 和路由在制作者启动期间启动，并导致路由启动失败。通过将这个启动延迟到延迟，那么可以通过 Camel 的路由错误处理程序路由消息期间，可以处理启动失败。请注意，在处理第一个消息时，创建并启动制作者可能需要稍等时间，并延长处理的总处理时间。

false

布尔值

lingerMs (producer)

生产者组将请求传输到单个批处理请求之间的任何记录组合在一起。通常，只有在记录到达的速度超过发送时，才会在负载下进行。然而，在某些情况下，客户端可能需要减少请求数量，即使是中等负载。此设置通过添加少量人工延迟来实现此目的，而不是立即发送一个记录者将等待给定的延迟，以允许发送其他记录。这可以被认为与 TCP 中的 Nagle 的算法类似。此设置给出了批处理延迟的上限：当我们获得批处理时，无论此设置如何立即发送分区，则设置将立即发送，但如果我们为这个分区有多个字节，我们将在等待更多记录进行"闲置"时显示。此设置默认为 0 （例如，没有延迟）。例如，设置 linger.ms=5 将具有减少发送的请求数量的影响，但会向负载发送的记录增加最多 5ms 的延迟。

0

整数

maxBlockMs (producer)

配置控制发送到 kafka 的块的时长。这些方法可能会因为多种原因而被阻止。对于 e.g: buffer full，元数据不可用。此配置对获取元数据、键和值序列化、分区和值在进行发送时分配缓冲区内存的总时间施加最大限制。如果是 partitionFor （），此配置在等待元数据时会实施最大时间阈值。

60000

整数

maxInFlightRequest (producer)

在阻塞前，客户端将在单个连接上发送的最大未确认请求数量。请注意，如果将此设置设定为大于 1，并且有失败发送，则会因为重试而重新排序消息的风险（例如，如果启用了重试）。

5

整数

maxRequestSize (producer)

请求的最大大小。这也是最大记录大小上限。请注意，服务器在记录大小上具有自己的 cap 值，可能不同于此大小。此设置将限制制作者在单个请求中发送的记录批处理数，以避免发送大型请求。

1048576

整数

metadataMaxAgeMs (producer)

以毫秒为单位的时间，我们强制刷新元数据，即使我们未看到任何分区领导更改来主动发现任何新的代理或分区。

300000

整数

metricReporters (producer)

用作指标报告器的类列表。通过实施 MetricReporter 接口，可以插入新指标创建通知的类。JmxReporter 始终包含在注册 JMX 统计数据中。

字符串

metricsSampleWindowMs (producer)

维护到计算指标的示例数量。

30000

整数

noOfMetricsSample (producer)

维护到计算指标的示例数量。

2

整数

Partitioner (producer)

在子主题之间分区信息的分区类。默认分区器基于密钥的哈希。

org.apache.kafka.clients.producer.internals.DefaultPartitioner

字符串

partitionKey (producer)

将向其发送记录的分区（如果没有指定分区，则为 null）。如果配置了这个选项，则它优先于标头 KafkaConstants#PARTITION_KEY。

整数

producerBatchSize (producer)

每当将多个记录发送到同一分区时，生产者都将尝试将记录一起批处理为更少的请求。这有助于客户端和服务器上的性能。此配置控制默认的批处理大小，以字节为单位。不尝试批处理记录大于这个大小。发送到代理的批处理将包含多个批处理，每个分区都有可用的数据。小型批处理大小将不太常见，并且可能会减少吞吐量（零的批处理大小将完全禁用批量）。非常大的批处理大小可能会更加浪费地使用内存，因为我们将始终在分析额外记录时分配指定批处理大小的缓冲区。

16384

整数

queueBufferingMaxMessages (producer)

在使用 async 模式时可以排队制作者时可以排队制作者的最大未发送消息数，然后才能被阻塞，或者必须丢弃数据。

10000

整数

receiveBufferBytes (producer)

读取数据时使用的 TCP 接收缓冲区的大小(SO_RCVBUF)。

65536

整数

reconnectBackoffMs (producer)

尝试重新连接到给定主机前等待的时间。这可避免在紧张循环中重复连接到主机。这个 backoff 适用于消费者向代理发送的所有请求。

50

整数

recordMetadata (producer)

制作者是否应该存储 RecordMetadata 结果到 Kafka。结果存储在包含 RecordMetadata 元数据的列表中。该列表存储在带有键 KafkaConstants#KAFKA_RECORDMETA 的标头中。

true

布尔值

requestRequiredAcks (producer)

在考虑请求完成前，生产者需要收到该制作者的确认数量。这控制发送的记录的持久性。以下设置很常见：acks=0 如果设为零，则制作者不会等待服务器中的任何确认。记录将立即添加到套接字缓冲区中，并被视为已发送。不保证服务器在这种情况下收到了记录，重试配置将无效（因为客户端通常不知道任何故障）。每个记录给出的偏移始终设为 -1。acks=1 意味着领导会将记录写入其本地日志，但不会等待所有后续者完全确认。在这种情况下，领导会在确认记录后立即失败，但在后续者复制该记录之前，记录将会丢失。acks=all 意味着领导机将等待一整组同步副本确认记录。这样可保证，只要至少有一个同步副本保持活跃，记录就不会被丢失。这是最强的可用保证。

Enum 值：

1

字符串

requestTimeoutMs (producer)

在向客户端发送错误前，代理会等待尝试满足 request.required.acks 要求的时间。

30000

整数

retries (producer)

设置大于零的值将导致客户端重新发送发送失败的任何记录，并具有潜在的瞬时错误。请注意，如果客户端在收到错误时重新显示记录，则这个重试不会不同。允许重试可能会更改记录的排序，因为如果两个记录被发送到一个分区，并且第一个记录失败，但重试，但第二次记录可能会首先出现。

0

整数

retryBackoffMs (producer)

在每个重试前，生产者会刷新相关主题的元数据，以查看是否已选择新的领导。由于领导选举需要一些时间，此属性指定制作者在刷新元数据前等待的时间。

100

整数

sendBufferBytes (producer)

套接字写入缓冲区大小。

131072

整数

valueSerializer (producer)

消息的序列化类。

org.apache.kafka.common.serialization.StringSerializer

字符串

Workerpool (producer)

要在 kafka 服务器确认消息从 KafkaProducer 使用异步非阻塞处理，使用自定义 worker 池继续路由 Exchange。如果使用这个选项，则必须处理线程池的生命周期，以便在需要时关闭池。

ExecutorService

workerPoolCoreSize (producer)

在 kafka 服务器确认从 KafkaProducer 使用异步非阻塞处理发送到 KafkaProducer 的消息后，worker 池的核心线程数量将继续路由 Exchange。

10

整数

workerPoolMaxSize (producer)

在 kafka 服务器确认从 KafkaProducer 使用异步非阻塞处理发送到 KafkaProducer 的消息后，worker 池的最大线程数量可以继续路由 Exchange。

20

整数

autowiredEnabled （高级）

是否启用自动关闭。这用于自动关闭选项（选项必须标记为 autowired），方法是在 registry 中查找查找是否有单个匹配类型实例，然后在组件上配置。这可以用于自动配置 JDBC 数据源、JMS 连接工厂、AWS 客户端等。

true

布尔值

kafkaClientFactory （高级）

Autowired Factory 用于创建 org.apache.kafka.clients.consumer.KafkaConsumer 和 org.apache.kafka.clients.producer.KafkaProducer 实例。这允许配置自定义工厂，以使用扩展 vanilla Kafka 客户端的逻辑创建实例。

KafkaClientFactory

同步 （高级）

设置是否应严格使用同步处理。

false

布尔值

schemaRegistryURL (confluent)

要使用的 Confluent Platform 模式 registry 服务器的 URL。格式为 host1:port1,host2:port2。这在 Confluent Platform 文档中被称为 schema.registry.url。这个选项只在 Confluent Platform （不是标准 Apache Kafka）中可用。

字符串

interceptorClasses (monitoring)

为生产者或消费者设置拦截器。生产者拦截器必须作为类实施 org.apache.kafka.clients.producer.ProducerInterceptor Consumer interceptors 必须是实施 org.apache.kafka.clients.consumer.ConsumerInterceptor 的类，如果您对消费者使用 Producer 拦截器，它将在运行时抛出一个类广播异常。

字符串

kerberosBeforeReloginMinTime (security)

刷新尝试之间的登录线程睡眠时间。

60000

整数

kerberosInitCmd (security)

Kerberos kinit 命令路径。默认为 /usr/bin/kinit。

/usr/bin/kinit

字符串

kerberosPrincipalToLocalRules (security)

从主体名称映射到短名称（通常是操作系统用户名）的规则列表。规则按顺序评估，第一个匹配主体名称的规则用于将其映射到短名称。列表中的任何后续规则都会被忽略。默认情况下，{username}/{hostname}{REALM} 形式的主体名称映射到 {username}。有关格式的详情，请查看安全授权和 acls 文档。可以使用逗号分隔多个值。

DEFAULT

字符串

kerberosRenewJitter (security)

添加到续订时间的随机 jitter 百分比。

0.05

双

kerberosRenewWindowFactor (security)

登录线程会处于睡眠状态，直到最后一次刷新到票据到期时间的指定窗口因子被达到，此时它将尝试续订票据。

0.8

双

saslJaasConfig (security)

公开 kafka sasl.jaas.config 参数示例： org.apache.kafka.common.security.plain.PlainLoginModule 需要 username=USERNAME password=PASSWORD;。

字符串

saslKerberosServiceName (security)

Kafka 运行的 Kerberos 主体名称。这可以在 Kafka 的 JAAS 配置或 Kafka 配置中定义。

字符串

saslMechanism (security)

使用简单身份验证和安全层(SASL)机制。有关有效值，请参阅。

GSSAPI

字符串

securityProtocol (security)

用于与代理通信的协议。支持 SASL_PLAINTEXT、PLAINTEXT 和 SSL。

PLAINTEXT

字符串

sslCipherSuites (security)

密码套件列表。这是身份验证、加密、MAC 和密钥交换算法的命名组合，用于使用 TLS 或 SSL 网络协议协商网络连接的安全设置。By 默认支持所有可用的密码套件。

字符串

sslContextParameters (security)

使用 Camel SSLContextParameters 对象的 SSL 配置.如果在其他 SSL 端点参数之前已进行了配置，则应用它。注意：Kafka 只支持从文件位置加载密钥存储，以便在 KeyStoreParameters.resource 选项中使用 file: 前缀。

SSLContextParameters

sslEnabledProtocols (security)

为 SSL 连接启用的协议列表。TLSv1.2、TLSv1.1 和 TLSv1 默认启用。

字符串

sslEndpointAlgorithm (security)

使用服务器证书验证服务器主机名的端点标识算法。

https

字符串

sslKeymanagerAlgorithm (security)

用于 SSL 连接的密钥管理器工厂使用的算法。Default 值是为 Java 虚拟机配置的密钥管理器工厂算法。

SunX509

字符串

sslKeyPassword (security)

密钥存储文件中私钥的密码。这对客户端是可选的。

字符串

sslKeystoreLocation (security)

密钥存储文件的位置。这是可选的，可用于客户端双向身份验证。

字符串

sslKeystorePassword (security)

密钥存储文件的存储密码。这对客户端是可选的，只有在配置了 ssl.keystore.location 时才需要。

字符串

sslKeystoreType (security)

密钥存储文件的文件格式。这对客户端是可选的。默认值为 JKS。

JKS

字符串

SSLProtocol ( 安全)

用于生成 SSLContext 的 SSL 协议。默认设置为 TLS，对于大多数情况来说都可以正常工作。最近的 JVM 中允许的值是 TLS、TLSv1.1 和 TLSv1.2。较旧的 JVM 中可能支持 SSL、SSLv2 和 SSLv3，但由于已知的安全漏洞，不建议使用它们。

字符串

sslProvider （安全）

用于 SSL 连接的安全提供程序的名称。默认值为 JVM 的默认安全提供程序。

字符串

sslTrustmanagerAlgorithm (security)

信任管理器工厂用于 SSL 连接的算法。Default 值是为 Java 虚拟机配置的信任管理器工厂算法。

PKIX

字符串

sslTruststoreLocation (security)

信任存储文件的位置。

字符串

sslTruststorePassword (security)

信任存储文件的密码。

字符串

sslTruststoreType (security)

信任存储文件的文件格式。默认值为 JKS。

JKS

字符串

useGlobalSslContextParameters (security)

启用对全局 SSL 上下文参数的使用。

false

布尔值