Kafka——Producer/Consumer
Kafka
- 特点：高吞吐、低延迟
Producer（生产者）
- 定义：是向Kafka集群发送消息的客户端应用，如：日志收集器、传感器数据发送器等
一、关键特性
- 异步发送：可提高吞吐量和降低延迟
- 消息分区：进而实现负载均衡
- 消息压缩：可使用gizp、snappy等压缩算法实现消息压缩，然后再传输到服务器
- 批处理机制：可减少网络I/O操作次数，提高总体的消息吞吐量
二、优化
- 发送缓冲区：在异步发送消息时，可以配置发送缓冲区来调整发送性能。通过调整发送缓冲区的大小，可以平衡性能和内存使用
- 重试机制：难免出现发送消息失败的情况，建议考虑设置重试机制，确保消息的可靠性
- 批处理大小和延迟：调整此参数可优化发送性能。较大的批处理大小和较长的延迟可减少网络开销和提高吞吐量
三、配置项
- bootstrap.servers：指定Kafka集群所需的broker地址清单，用于建立初始连接
- acks：指定分区中必须有多少个副本收到这条消息，才算消息发送成功。可选值：
- 0（不等待任何相应）
- 1（leader副本确认）
- all/-1（ISR中所有副本确认）
- retries：消息发送失败时的重试次数，用于处理临时故障
- compression.type：消息压缩算法，用于减少网络传输量。可选值：
- none
- gzip
- snappy
- ls4
- zstd
- batch.size：每个批次的内存大小，影响消息的吞吐量和延迟
- linger.ms：发送批次前的最大等待时间，用于平衡吞吐量和延迟
- buffer.memory：生产者客户端中用于缓存消息的缓冲区大小
- key.serializer和value.serializer：消息的键和值的序列化类，用于将消息转换为字节数组
- max.request.size:生产者客户端能发送的最大消息大小
- request.timeout.ms：生产者等待请求响应的最长时间
四、常用方法
send()：
定义：发送消息到kafka集群
- 异步发送：直接调用
send(record)，不关心发送结果（但可通过回调函数处理结果）。 - 同步发送：调用
send(record).get()，等待Kafka响应，获取发送结果（RecordMetadata对象）。 - 带回调的异步发送：调用
send(record, new Callback() {...})，在回调函数中处理发送结果。
- 异步发送：直接调用
close()：
定义关闭生产者，释放资源
Consumer（消费者）
- 定义：从kafka集群读取消息的客户端应用。如：数据分析工具、实时监控系统等
一、关键特性
- 拉取模式
- 消费者组：消费组->消费主题->消费分区，从而实现负载均衡。不同消费者组之间互不影响，可以独立地消费同一主题的消息
- 偏移量管理：消费者可维护偏移量（Offset）来记录自己再分区中消费的位置。偏移量记录方式：重置到较旧的偏移量以重新处理过去的消息；跳到最近的记录从“现在”开始记录
二、优化
- 自动提交偏移量：简化了消费者的实现，但可能导致消费消息重复和丢失。手动提交偏移量提供了更灵活的控制方式，适合需要确保消息消费成功的场景
- 消费速率控制：消息者处理速度较慢，可能导致消息积压。即可以通过增加消费者实例、优化消费者逻辑等方式来解决问题
- 错误处理与重试：遇到的错误如：网络错误、消息格式错误等。消费者需实现适当的错误处理机制，如重试、记录错误日志等来确保消息的可靠性
三、配置项
- bootstrap.servers：同样用于指定kafka集群的broker地址清单
- group.id：消费者所属消费组的唯一标识，用于协调消费者之间的消息分配
- auto.offset.reset：当分区没有初始偏移量或偏移量无效时，消费者的行为。可选值：earliest（从最早的消息开始消费）、latest（从最新的消息开始消费）和none（抛出异常）
- enable.auto.commit：是否开启自动提交消费位移的功能。
- true（定期自动提交消费位移）
- false（需要手动提交）
- anto.commit.interval.ms：自动提交消费位移的时间间隔
- fetch.min.bytes：消费者客户端一次请求从kafka拉取消息的最小数据量
- fetch.max.bytes：消费者客户端一次请求从kafka拉取消息的最大数据量
- max.poll.records：一次拉取请求的最大消息数
- max.poll.interval.ms：指定拉取消息线程最长空闲时间
- session.timeout.ms：检测消费者是否失效的超时时间
- heartbeat.interval.ms：消费者心跳时间，用于维持与消费者协调器的连接
- key.deserializer和value.deserializer：消息的键和值的反序列化类，用于将字节数据转换为消息对象
四、常用方法
poll()
含义：消费者向kafka服务器发起一次拉取请求
参数
- 超时时间
- 含义：控制消费者在单次poll循环中，如果没拿到数据，最多能等多久
- 目的：避免忙循环、控制循环速度
subscribe()
- 用于订阅一个或多个主题
commitSync()或commitAsync()
- 手动提交偏移量
java和python开发的区别
1、Java
- consumer-poll()方法
- 返回结构：一组消息的集合-ConsumerRecords对象
- 优点：提高吞吐量，批量效率更高
2、python
- consumer-poll()方法
- 返回结构：单条消息或None，但内部有批量预取机制
- 批量预取机制：在后台批量拉取消息存在内部缓冲区中，当循环调用poll()时，从内部缓冲区依次取出单条消息，也相当于一次拉取了多条消息