0


kafka集成篇

kafka的Java客户端

生产者

1.引入依赖

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.6.3</version></dependency>

2.生产者发送消息的基本实现

/**
 * 消息的发送⽅
 */publicclassMyProducer{privatefinalstaticStringTOPIC_NAME="my-replicated-topic";publicstaticvoidmain(String[] args){Properties props =newProperties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"124.222.253.33:9092,124.222.253.33:9093,124.222.253.33:9094");// 把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// 把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());RecordMetadata metadata =null;try(Producer<String,String> producer =newKafkaProducer<>(props)){Order order =newOrder(1L,99.9D);// 未指定发送分区,具体发送的分区计算公式:hash(key)%partitionNumProducerRecord<String,String> producerRecord =newProducerRecord<>(TOPIC_NAME, order.getOrderId().toString(),JSON.toJSONString(order));// 等待消息发送成功的同步阻塞⽅法
            metadata = producer.send(producerRecord).get();}catch(InterruptedException|ExecutionException e){thrownewRuntimeException(e);}finally{if(metadata !=null){// =====阻塞=======System.out.println("同步⽅式发送消息结果:"+"topic-"+
                        metadata.topic()+"|partition-"+ metadata.partition()+"|offset-"+
                        metadata.offset());}}}}

3.发送消息到指定分区

image-20230814170847906

4.发送消息未指定分区

发送消息未指定分区,会通过业务key的hash运算,算出消息往哪个分区上发

// 未指定发送分区,具体发送的分区计算公式:hash(key)%partitionNumProducerRecord<String,String> producerRecord =newProducerRecord<>(TOPIC_NAME, order.getOrderId().toString(),JSON.toJSONString(order));

5.同步发送消息

image-20230814172356139

如果生产者发送消息没有收到ack,生产者会阻塞,阻塞到3s的时间,如果还没有收到消息,会进行重试。重试的次数3次。

RecordMetadata metadata = producer.send(producerRecord).get();System.out.println("同步⽅式发送消息结果:"+"topic-"+
    metadata.topic()+"|partition-"+ metadata.partition()+"|offset-"+ metadata.offset());

6.异步发送消息

image-20230814173250894

异步发送,生产者发送完消息后就可以执行之后的业务,broker在收到消息后异步调用生产者提供的callback回调方法。

// 异步发送消息 Callback回调接口
            producer.send(producerRecord,newCallback(){// 异步回调方法@OverridepublicvoidonCompletion(RecordMetadata metadata,Exception e){if(e !=null){System.err.println("发送消息失败:"+
                                e.getMessage());}if(metadata !=null){System.out.println("异步⽅式发送消息结果:"+"topic-"+
                                metadata.topic()+"|partition-"+ metadata.partition()+"|offset-"+ metadata.offset());}}});System.out.println("处理之后的逻辑~");

输出结果:

image-20230814173709486

7.生产者中的ack的配置

在同步发消息的场景下:生产者发送消息到broker上后,ack会有3种不同的选择

  • ack = 0 :kafka-cluster不需要任何的broker收到消息,就立即返回ack给生产者就可以继续发送下一条消息,效率是最高的但最容易丢消息
  • ack=1(默认):多副本之间的leader已经收到消息,并把消息写⼊到本地的log中,才会返回ack给生产者,性能和安全性是最均衡的(这种情况下,如果follower没有成功备份数据,而此时leader又挂掉,则消息会丢失)
  • ack=-1/all:需要等待 min.insync.replicas(默认为1,推荐配置大于等于2) 这个参数配置的副本个数都成功写入日志才会返回ack给生产者,这种策略会保证只要有⼀个备份存活就不会丢失数据。这种方式最安全但性能最差。(⼀般除非是金融级别,或跟钱打交道的场景才会使用这种配置)

image-20230814175916962

code:

props.put(ProducerConfig.ACKS_CONFIG,"1");

关于ack和重试(如果没有收到ack,就开启重试)的配置

  • 发送会默认会重试3次,每次间隔100ms
props.put(ProducerConfig.ACKS_CONFIG,"1");/*
 发送失败会重试,默认重试间隔100ms,【重试能保证消息发送的可靠性,但是也可能造成消息重复发送】,⽐如⽹络抖动,所以【需要在接收者那边做好消息接收的幂等性处理】
 */
 props.put(ProducerConfig.RETRIES_CONFIG,3);// 重试间隔设置
 props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,300);

8.关于消息发送的缓冲区

发送的消息会先进入到本地缓冲区(32mb),kakfa会跑⼀个线程,该线程去缓冲区中取16k的数据,发送到kafka,如果到10毫秒数据没取满16k,也会发送⼀次。

image-20230814180715793

  • kafka默认会创建一个消息缓冲区,用来存放要发送的消息,缓冲区是32m
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
  • kafka本地线程会去缓冲区中⼀次拉16k的数据,发送到broker
props.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
  • 如果线程拉不到16k的数据,间隔10ms也会将已拉到的数据发到broker
props.put(ProducerConfig.LINGER_MS_CONFIG,10);

消费者

1.消费者消费消息的基本实现

publicclassMyConsumer{privatefinalstaticStringTOPIC_NAME="my-replicated-topic";privatefinalstaticStringCONSUMER_GROUP_NAME="testGroup";publicstaticvoidmain(String[] args){Properties props =newProperties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"124.222.253.33:9092,124.222.253.33:9093,124.222.253.33:9094");// 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG,CONSUMER_GROUP_NAME);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());// 1.创建⼀个消费者的客户端try(KafkaConsumer<String,String> consumer =newKafkaConsumer<>(props)){// 2.消费者订阅主题列表
            consumer.subscribe(Collections.singletonList(TOPIC_NAME));while(true){/*
                 * 3.poll()API 是拉取消息的⻓轮询
                 */ConsumerRecords<String,String> records =
                        consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String> record : records){// 4.操作消息System.out.printf("收到消息:partition = %d,offset = %d, key = %s, value = %s%n ", record.partition(), record.offset(), record.key(), record.value());}}}catch(Exception e){thrownewRuntimeException(e);}}}

2.消费者自动提交和手动提交offset

1)提交的内容

消费者无论是自动提交还是手动提交,都需要把所属的消费组+消费的某个主题+消费的某个分区及消费的偏移量,这样的信息提交到集群的_consumer_offsets主题里面。

2)自动提交

消费者poll消息下来以后就会自动提交offset

// 是否自动提交offset,默认就是true
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");// 自动提交offset的间隔时间
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
注意

自动提交会丢消息。因为消费者在消费前提交offset,有可能提交完后还没消费时消费者挂了。于是下⼀个消费者会从已提交的offset的下一个位置开始消费消息。之前未被消费的消息就丢失掉了。

3)手动提交

需要把自动提交的配置改成false

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
手动提交又分成了两种

  • 手动同步提交

在消费完消息后调用同步提交的方法,当集群返回ack前⼀直阻塞,返回ack后表示提交成功,执行之后的逻辑

while(true){/*
                 * poll()API 是拉取消息的⻓轮询
                 */ConsumerRecords<String,String> records =
                        consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String> record : records){// 操作消息System.out.printf("收到消息:partition = %d,offset = %d, key = %s, value = %s%n ", record.partition(), record.offset(), record.key(), record.value());}// 所有的消息已消费完if(records.count()>0){// 有消息// ⼿动同步提交offset,当前线程会阻塞直到offset提交成功// 【⼀般使⽤同步提交】,因为提交之后⼀般也没有什么逻辑代码了
                    consumer.commitSync();// =======阻塞=== 提交成功}}
  • 手动异步提交

在消息消费完后提交,不需要等到集群ack,直接执行之后的逻辑,可以设置⼀个回调方法,供集群调用

while(true){/*
                 * poll()API 是拉取消息的⻓轮询
                 */ConsumerRecords<String,String> records =
                        consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String> record : records){// 操作消息System.out.printf("收到消息:partition = %d,offset = %d, key = %s, value = %s%n ", record.partition(), record.offset(), record.key(), record.value());}// 所有的消息已消费完if(records.count()>0){// 有消息// ⼿动异步提交offset,当前线程提交offset不会阻塞,可以继续处理后⾯的程序逻辑
                    consumer.commitAsync(newOffsetCommitCallback(){@OverridepublicvoidonComplete(Map<TopicPartition,OffsetAndMetadata> offsets,Exception exception){if(exception !=null){System.err.println("Commit failed for "+ offsets);System.err.println("Commit failed exception: "+ exception.getMessage());}}});}}

3.长轮询poll消息(消费者拉取消息)

  • 消费者建立了与broker之间的长连接,开始poll消息
  • 默认情况下,消费者一次会poll500条消息
// ⼀次poll最⼤拉取消息的条数,可以根据消费速度的快慢来设置
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,500);
  • 代码中设置了长轮询的时间是1000毫秒
while(true){/*
                 * poll()API 是拉取消息的⻓轮询
                 */ConsumerRecords<String,String> records =
                        consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String> record : records){System.out.printf("收到消息:partition = %d,offset = %d, key = %s, value = %s%n ", record.partition(), record.offset(), record.key(), record.value());}}
  • 意味着: - 如果⼀次poll到500条,就直接执行for循环- 如果这⼀次没有poll到500条。且时间在1秒内,那么长轮询继续poll,要么到500条,要么到1s,执行后续for循环- 如果多次poll都没达到500条,且1秒时间到了,那么直接执行for循环- 如果两次poll的间隔超过30s(poll时间短但是消费时间长,消费者消费可能会达到30s左右),集群会认为该消费者的消费能力过 弱,该消费者被踢出消费组,触发rebalance机制,rebalance机制会造成性能开销

可以通过设置参数, 让⼀次poll的消息条数少⼀点,避免触发rebalance损耗性能

// ⼀次poll最⼤拉取消息的条数,可以根据消费速度的快慢来设置
 props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,500);// 如果两次poll的时间如果超出了30s的时间间隔,kafka会认为其消费能⼒过弱,将其踢出消费组。将分区分配给其他消费者。-rebalance
 props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,30*1000);

4.消费者的健康状态检查

消费者每隔1s向kafka集群发送心跳,集群发现如果有超过10s没有续约的消费者,将被踢出消费组,触发该消费组的rebalance机制,将该分区交给消费组里的其他消费者进行消费。

// consumer给broker发送心跳的间隔时间  1s一次
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,1000);// kafka如果超过10秒没有收到消费者的心跳,则会把消费者踢出消费组,进⾏rebalance,把分区分配给其他消费者。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,10*1000);

5.指定分区和偏移量、时间消费

  • 指定分区消费
consumer.assign(Arrays.asList(newTopicPartition(TOPIC_NAME,0)));
  • 从头消费
consumer.assign(Arrays.asList(newTopicPartition(TOPIC_NAME,0)));
consumer.seekToBeginning(Arrays.asList(newTopicPartition(TOPIC_NAME,0)));
  • 指定offset消费
consumer.assign(Arrays.asList(newTopicPartition(TOPIC_NAME,0)));
consumer.seek(newTopicPartition(TOPIC_NAME,0),10);
  • 指定时间消费

根据时间,去所有的partition中确定该时间对应的offset,然后去所有的partition中找到该offset之后的消息开始消费。

// topic对应所有分区List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);// 从1小时前开始消费long fetchDataTime =newDate().getTime()-1000*60*60;Map<TopicPartition,Long> map =newHashMap<>();for(PartitionInfo par : topicPartitions){
    map.put(newTopicPartition(TOPIC_NAME, par.partition()), fetchDataTime);}Map<TopicPartition,OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);for(Map.Entry<TopicPartition,OffsetAndTimestamp> entry : parMap.entrySet()){TopicPartition key = entry.getKey();OffsetAndTimestamp value = entry.getValue();if(key ==null|| value ==null)continue;long offset = value.offset();System.out.println("partition-"+ key.partition()+"|offset-"+ offset);System.out.println();//根据消费⾥的timestamp确定offset
    consumer.assign(Arrays.asList(key));
    consumer.seek(key, offset);}

6.新消费组的消费offset规则

新消费组中的消费者在启动以后,默认会从当前分区的最后⼀条消息的offset+1开始消费(消费新消息)。可以通过以下的设置,让新的消费者第⼀次从头开始消费。之后开始消费新消息(最后消费的位置的偏移量+1)

  • Latest:默认的,消费新消息
  • earliest:第⼀次从头开始消费。之后开始消费新消息(最后消费的位置的偏移量+1),这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

SpringBoot集成kafka

1.引入依赖

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

2.配置文件

server:port:8080spring:kafka:bootstrap-servers: 124.222.253.33:9092,124.222.253.33:9093,124.222.253.33:9094producer:# 生产者retries:3# 设置大于0的值,则客户端会将发送失败的记录重新发送batch-size:16384# 每次拉取多少数据发送broker buffer-memory:33554432# 本地缓冲区大小acks:1# 指定消息key和消息体的编解码⽅式key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:group-id: default-group
      enable-auto-commit:falseauto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      max-poll-records:500listener:# 当每⼀条记录被消费者监听器(ListenerConsumer)处理之后提交# RECORD# 当每⼀批poll()的数据被消费者监听器(ListenerConsumer)处理之后提交# BATCH# 当每⼀批poll()的数据被消费者监听器(ListenerConsumer)处理之后,距离上次提交时间大于TIME时提交# TIME# 当每⼀批poll()的数据被消费者监听器(ListenerConsumer)处理之后,被处理record数量大于等于COUNT时提交# COUNT# TIME | COUNT 有⼀个条件满足时提交# COUNT_TIME# 当每⼀批poll()的数据被消费者监听器(ListenerConsumer)处理之后, 手动调用Acknowledgment.acknowledge()后提交# MANUAL# 【手动调用Acknowledgment.acknowledge()后立即提交,⼀般使用这种】# MANUAL_IMMEDIATEack-mode: MANUAL_IMMEDIATE

3.消息生产者

发送消息到指定topic

image-20230815110653415

4.消息消费者

设置消费组,消费指定topic

@ComponentpublicclassMyConsumer{@KafkaListener(topics ="my-replicated-topic", groupId ="MyGroup1")publicvoidlistenGroup(ConsumerRecord<String,String> record,Acknowledgment ack){String value = record.value();System.out.println(record);System.out.println(value);//⼿动提交offset
        ack.acknowledge();}}

5.消费者中配置消费主题、分区和偏移量

设置消费组、多topic、指定分区、指定偏移量消费及设置消费者个数

@KafkaListener(groupId ="testGroup", topicPartitions ={@TopicPartition(topic ="topic1", partitions ={"0","1"}),@TopicPartition(topic ="topic2", partitions ="0",
                    partitionOffsets =@PartitionOffset(partition ="1", initialOffset ="100"))}, concurrency ="3")// concurrency:同消费组中消费者个数,就是并发消费数,建议小于等于分区总数publicvoidlistenGroupPro(ConsumerRecord<String,String> record,Acknowledgment ack){String value = record.value();System.out.println(value);System.out.println(record);//⼿动提交offset
        ack.acknowledge();}
标签: kafka 分布式

本文转载自: https://blog.csdn.net/qq_43417581/article/details/132295560
版权归原作者 Likelong~ 所有, 如有侵权,请联系我们删除。

“kafka集成篇”的评论:

还没有评论