【Kafka】Consumer配置

简介:

从0.9.0.0开始,下面是消费者的配置。

名称 描述 类型 默认值
bootstrap.servers 消费者初始连接kafka集群时的地址列表。不管这边配置的什么地址,消费者会使用所有的kafka集群服务器。消费者会通过这些地址列表,找到所有的kafka集群机器。 list
key.deserializer 实现了Deserializer的key的反序列化类 class
value.deserializer 实现了Deserializer的value的反序列化类 class
fetch.min.bytes 每次请求,kafka返回的最小的数据量。如果数据量不够,这个请求会等待,直到数据量到达最小指标时,才会返回给消费者。如果设置大于1,会提高kafka的吞吐量,但是会有额外的等待期的代价。 int 1
group.id 标识这台消费者属于那个消费组。如果消费者通过订阅主题来实现组管理功能,或者使用基于kafka的偏移量管理策略,这个配置是必须的。 string ""
heartbeat.interval.ms 使用kafka集群管理工具时,消费者协调器之间的预计心跳时间。心跳的作用是确保消费者的session是活跃的,同时当新的机器加入集群或有机器挂掉的情况下触发再平衡操作。这个配置必须小于heartbeat.interval.ms,而且应该不大于这个值的1/3。为了控制正常的负载均衡的预期时间,这个值可以设置的更小。 int 3000
max.partition.fetch.bytes kafka集群每个分区一次返回的最大数据量。一次请求的最大内存使用量应该等于#partitions * max.partition.fetch.bytes。这个值必须与kafka集群允许的最大消息数据量差不多大小,否则可能生产者发送了一个消息,大于消费者配置的值。这种情况下,消费者可能会在获取那条消息时堵住。 int 1048576
session.timeout.ms 使用kafka集群管理工具时检测失败的超时时间。如果在session超时时间范围内,没有收到消费者的心跳,broker会把这个消费者置为失效,并触发消费者负载均衡。因为只有在调用poll方法时才会发送心跳,更大的session超时时间允许消费者在poll循环周期内处理消息内容,尽管这会有花费更长时间检测失效的代价。如果想控制消费者处理消息的时间,还可以参考max.poll.records。注意这个值的大小应该在group.min.session.timeout.ms和group.max.session.timeout.ms范围内。 int 30000
ssl.key.password 私钥存储文件的私钥密码。可选配置。 password null
ssl.keystore.location 私钥存储文件的路径。可选配置,并且可用来作为客户端的双向认证。 string null
ssl.keystore.password 私钥存储文件的存储密码。可选配置,并且只有ssl.keystore.location配置的情况下才需要配置。 password null
ssl.truststore.location 信任秘钥文件路径。 string null
ssl.truststore.password 信任秘钥文件密码。 password null
auto.offset.reset 当kafka的初始偏移量没了,或者当前的偏移量不存在的情况下,应该怎么办?下面有几种策略:earliest(将偏移量自动重置为最初的值)、latest(自动将偏移量置为最新的值)、none(如果在消费者组中没有发现前一个偏移量,就向消费者抛出一个异常)、anything else(向消费者抛出异常) string latest
connections.max.idle.ms 配置时间后,关闭空闲的连接 long 540000
enable.auto.commit 如果设为true,消费者的偏移量会定期在后台提交。 boolean true
exclude.internal.topics 内部主题(比如偏移量)是否需要暴露给消费者。如果设为true,获取内部主题消息的途径就是订阅他们。 boolean true
max.poll.records 一次poll调用返回的最大消息数量。 int 2147483647
partition.assignment.strategy 使用组管理时,客户端使用的分区策略的类名,根据这个策略来进行消费分区。 list [org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes SO_RCVBUF读取数据使用的内存大小。 int 65536
request.timeout.ms 这个配置控制一次请求响应的最长等待时间。如果在超时时间内未得到响应,kafka要么重发这条消息,要么超过重试次数的情况下直接置为失败。 int 40000
目录
相关文章
|
18天前
|
消息中间件 存储 Prometheus
Kafka集群如何配置高可用性
Kafka集群如何配置高可用性
|
1月前
|
消息中间件 监控 Ubuntu
大数据-54 Kafka 安装配置 环境变量配置 启动服务 Ubuntu配置 ZooKeeper
大数据-54 Kafka 安装配置 环境变量配置 启动服务 Ubuntu配置 ZooKeeper
75 3
大数据-54 Kafka 安装配置 环境变量配置 启动服务 Ubuntu配置 ZooKeeper
|
1月前
|
消息中间件 存储 分布式计算
大数据-53 Kafka 基本架构核心概念 Producer Consumer Broker Topic Partition Offset 基础概念了解
大数据-53 Kafka 基本架构核心概念 Producer Consumer Broker Topic Partition Offset 基础概念了解
63 4
|
1月前
|
消息中间件 分布式计算 Java
大数据-73 Kafka 高级特性 稳定性-事务 相关配置 事务操作Java 幂等性 仅一次发送
大数据-73 Kafka 高级特性 稳定性-事务 相关配置 事务操作Java 幂等性 仅一次发送
31 2
|
1月前
|
消息中间件 Java 大数据
大数据-56 Kafka SpringBoot与Kafka 基础简单配置和使用 Java代码 POM文件
大数据-56 Kafka SpringBoot与Kafka 基础简单配置和使用 Java代码 POM文件
65 2
|
1月前
|
消息中间件 NoSQL Kafka
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
130 0
|
2月前
|
消息中间件 安全 大数据
Kafka多线程Consumer是实现高并发数据处理的有效手段之一
【9月更文挑战第2天】Kafka多线程Consumer是实现高并发数据处理的有效手段之一
250 4
|
3月前
|
消息中间件 大数据 Kafka
Kafka消息封装揭秘:从Producer到Consumer,一文掌握高效传输的秘诀!
【8月更文挑战第24天】在分布式消息队列领域,Apache Kafka因其实现的高吞吐量、良好的可扩展性和数据持久性备受开发者青睐。Kafka中的消息以Record形式存在,包括固定的头部与可变长度的消息体。生产者(Producer)将消息封装为`ProducerRecord`对象后发送;消费者(Consumer)则从Broker拉取并解析为`ConsumerRecord`。消息格式简化示意如下:消息头 + 键长度 + 键 + 值长度 + 值。键和值均为字节数组,需使用特定的序列化/反序列化器。理解Kafka的消息封装机制对于实现高效、可靠的数据传输至关重要。
82 4
|
3月前
|
消息中间件 监控 Java
【Kafka节点存活大揭秘】如何让Kafka集群时刻保持“心跳”?探索Broker、Producer和Consumer的生死关头!
【8月更文挑战第24天】在分布式系统如Apache Kafka中,确保节点的健康运行至关重要。Kafka通过Broker、Producer及Consumer间的交互实现这一目标。文章介绍Kafka如何监测节点活性,包括心跳机制、会话超时与故障转移策略。示例Java代码展示了Producer如何通过定期发送心跳维持与Broker的连接。合理配置这些机制能有效保障Kafka集群的稳定与高效运行。
73 2