上市公司网站建设分析评价,装修公司哪家好排名,关于网站建设的网站,wordpress文章加密提示客户端
api
Kafka提供了以下两套客户端API
HighLevel(重点)LowLevel
HighLevel API封装了kafka的运行细节#xff0c;使用起来比较简单#xff0c;是企业开发过程中最常用的客户端API。 而LowLevel API则需要客户端自己管理Kafka的运行细节#xff0c;Partition#x…客户端
api
Kafka提供了以下两套客户端API
HighLevel(重点)LowLevel
HighLevel API封装了kafka的运行细节使用起来比较简单是企业开发过程中最常用的客户端API。 而LowLevel API则需要客户端自己管理Kafka的运行细节PartitionOffset这些数据都由客户端自行管理。这层API功能更灵活但是使用起来非常复杂也更容易出错。只在极少数对性能要求非常极致的场景才会偶尔使用 生产者发送消息
发送流程:
组装生产者核心配置参数初始化生产者组装消息发送消息, 三种模式 单向发送, 不等待broker返回结果同步发送异步发送关闭生产者 代码:
package com.kk.kafka.demo;import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;public class ProducerTest {public static final String KAFKA_URL 192.168.6.128:9092;public static final String TOPIC oneTopic;public static void main(String[] args) throws ExecutionException, InterruptedException {// 组装生产者配置Properties ps new Properties();ps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_URL);ps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);ps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);// 初始化生产者ProducerString, String producer new KafkaProducer(ps);for (int i 0; i 5; i) {ProducerRecordString, String producerRecord new ProducerRecord(TOPIC, key i, message i);// 同步发送producer.send(producerRecord);// 同步发送RecordMetadata metadata producer.send(producerRecord).get();//异步发送producer.send(producerRecord, new Callback() {Overridepublic void onCompletion(RecordMetadata metadata, Exception e) {if (metadata ! null) {System.out.println(Message sent successfully! Topic: metadata.topic() , Partition: metadata.partition() , Offset: metadata.offset() , message: producerRecord.value());} else {System.err.println(Error sending message: e.getMessage());}}});}producer.close();}
} 消费者消费消息
消费流程:
组装消费者核心配置参数初始化消费者订阅topic, 可订阅多个拉取消息, 可配置超时时间提交offset, 分为同步和异步两种方式, 服务端维护offset消费进度 代码:
package com.kk.kafka.demo;import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;public class ConsumerTest {public static final String KAFKA_URL 192.168.6.128:9092;public static final String TOPIC oneTopic;public static void main(String[] args) {// 组装消费者配置参数Properties props new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_URL);props.put(ConsumerConfig.GROUP_ID_CONFIG, your-consumer-group);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);// 初始化消费者ConsumerString, String consumer new KafkaConsumer(props);// 订阅topicconsumer.subscribe(Arrays.asList(TOPIC));while (true) {// 拉取消息, 100毫秒超时时间ConsumerRecordsString, String records consumer.poll(Duration.ofNanos(100));//处理消息for (ConsumerRecordString, String record : records) {System.out.println(start Consumer offset record.offset() ;key record.key() ; value record.value());}//提交offset消息就不会重复推送。//同步提交表示必须等到offset提交完毕再去消费下一批数据。consumer.commitSync();//异步提交表示发送完提交offset请求后就开始消费下一批数据了。不用等到Broker的确认。
// consumer.commitAsync();}}
} 客户端整体流程 拦截器
序列化器
发送到Dequeue
Dequeue满了或者批次满了或者阈值时间 推到InflightRequest
send线程将InflightRequest推到服务端Partition, 满足一定阈值
缓存机制
broker给生产者ack 消费者分组策略
相关消费者配置
GROUP_ID_CONFIG: 群组唯一idGROUP_INSTANCE_ID_CONFIG: 消费实例id, 可以减少不必要rebalance
生产者给Topic投递消息, 消息会均匀的存到partition. 消息会向所有订阅该Topic的消费实例推送, 推送时, 一个消费群组只会推送一份. 也就是, 同一个群组里面, 只会有一个消费实例能消费; 不同消费群组可以重复消费消息.Offset就是记录每个消费群组在partition的处理消息进度 offset丢失怎么解决
1. offset丢失:
初始化consumer group时, 设置offset失败offset对应的数据文件被删除
解决: 服务端有个兜底方案, 可以配置消费者配置 ConsumerConfig.AUTO_OFFSET_RESEWT_CONFIG 当Server端没有对应的Offset时要如何处理。 可选项 earliest 自动设置为当前最早的offsetlatest自动设置为当前最晚的offsetnone 如果消费者组对应的offset找不到就向Consumer抛异常。其他选项 向Consumer抛异常。 2. offset不一致问题
消费者可以选择同步提交或者异步提交
同步提交: 消息处理完, 提交. 消息处理失败, 选择不提交, 等重试. 如果消费过慢, 服务端不会无限等, 会认为本次消费失败, 会给同组的其他消费实例投递消费, 可能存在重复消费问题 异步提交: 先提交, 后处理消息. 如果消息处理失败, offset又被提交, 就存在客户端与服务端offset不一致问题
解决: 可以交由客户端管理offset, 存到redis或者mysql等中间件.客户端可以控制消息处理进度, 实时推进offset 生产者拦截器
生产者配置: INTERCEPTOR_CLASSES_CONFIG properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, com.kk.kafka.demo.MyProducerInterceptor);public class MyProducerInterceptor implements ProducerInterceptor {// 发送消息触发Overridepublic ProducerRecord onSend(ProducerRecord producerRecord) {System.out.println(onSend producerRecord: producerRecord.toString());return producerRecord;}// 收到服务端相应触发Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {System.out.println(acknowledgement recordMetadata: recordMetadata.toString());}// 生产者连接关闭触发Overridepublic void close() {System.out.println(producer close);}// 整理配置项Overridepublic void configure(MapString, ? map) {}
} 消息序列化
相关配置:
KEY_SERIALIZER_CLASS_CONFIG: key序列号VALUE_SERIALIZER_CLASS_CONFIG: 消息value序列号KEY_DESERIALIZER_CLASS_CONFIG: key反序列化VALUE_DESERIALIZER_CLASS_CONFIG: 消息value反序列化
1. 生产者:
ps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);
ps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);
2. 消费者
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer); 消息分区路由机制 生产者缓存机制 其他重要核心参数 幂等性原理 事务消息
