kafka发送消息-生产者发送消息的分区计谋（消息发送到哪个分区中？是什么策 ...

乌市泽哥  金牌会员 | 2024-8-28 07:43:11 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 567|帖子 567|积分 1701


1、默认计谋，步伐自动计算并指定分区

1.1、指定key，不指定分区

生产者：在编写代码发送消息时我们先不指定分区，即分区设为null，看看步伐终极会把消息发送到哪个分区。
  1. package com.power.producer;
  3. import com.power.model.User;
  4. import org.apache.kafka.clients.producer.ProducerRecord;
  5. import org.apache.kafka.common.header.Headers;
  6. import org.apache.kafka.common.header.internals.RecordHeaders;
  7. import org.springframework.kafka.core.KafkaTemplate;
  8. import org.springframework.kafka.support.KafkaHeaders;
  9. import org.springframework.kafka.support.SendResult;
  10. import org.springframework.messaging.Message;
  11. import org.springframework.messaging.support.MessageBuilder;
  12. import org.springframework.stereotype.Component;
  13. import org.springframework.util.concurrent.ListenableFuture;
  15. import javax.annotation.Resource;
  16. import java.nio.charset.StandardCharsets;
  17. import java.util.Date;
  18. import java.util.concurrent.CompletableFuture;
  19. import java.util.concurrent.ExecutionException;
  21. @Component
  22. public class EventProducer {
  24.     @Resource
  25.     private KafkaTemplate<String,Object> kafkaTemplate2;
  26.    
  27.     public void send9(){
  28.         User user = User.builder().id(1208).phone("16767667676").birthday(new Date()).build();
  29.         //分区是null，让kafka自己去决定把消息发送到哪个分区
  30.         kafkaTemplate2.send("heTopic",null,System.currentTimeMillis(),"k9",user);
  31.     }
  32. }
复制代码
测试类：
  1. package com.power;
  3. import com.power.model.User;
  4. import com.power.producer.EventProducer;
  5. import org.junit.jupiter.api.Test;
  6. import org.springframework.boot.test.context.SpringBootTest;
  8. import javax.annotation.Resource;
  9. import java.util.Date;
  11. @SpringBootTest
  12. public class SpringBoot01KafkaBaseApplication {
  14.     @Resource
  15.     private EventProducer eventProducer;
  16.    
  17.     @Test
  18.     void send9(){
  19.         eventProducer.send9();
  20.     }
  21. }
复制代码
步伐终极是通过以下代码进行目标分区计算的：
  1. Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
复制代码
通过调试发现，步伐是通过以下代码进行目标分区计算的：
步伐自动读取生产者发送消息时的key（本次发送时值为“key9”），将key天生一个32位的HASH值，将该HASH值与默认分区数（这个topic中有9个分区）取余数（余数结果肯定在0-8之间），进而计算得出消息默认发送到的分区值


1.2、不指定key，不指定分区

生产者：

测试类：

此时时通过随机数与默认分区取余数计算默认分区的
  1. 使用随机数 % numPartitions
复制代码
2、轮询分配计谋RoundRobinPartitioner

通过查看kafka源码发现，分区接口有一个轮询分配计谋相关实现类。

在application.yml配置文件中生产者配置项，我发现并生产者并没有相关轮询分配计谋的配置，那么该如何试下轮询指定分区的配置呢？

须要编写代码试下轮询指定分区计谋：

2.1、创建配置类

  1. package com.power.config;
  3. import org.apache.kafka.clients.admin.NewTopic;
  4. import org.apache.kafka.clients.producer.ProducerConfig;
  5. import org.apache.kafka.clients.producer.RoundRobinPartitioner;
  6. import org.springframework.beans.factory.annotation.Value;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  9. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  10. import org.springframework.kafka.core.KafkaTemplate;
  11. import org.springframework.kafka.core.ProducerFactory;
  13. import java.util.HashMap;
  14. import java.util.Map;
  16. @Configuration
  17. public class KafkaConfig {
  19.     @Value("${spring.kafka.bootstrap-servers}")
  20.     private String bootstrapServers;
  22.     @Value("${spring.kafka.producer.key-serializer}")
  23.     private String keySerializer;
  25.     @Value("${spring.kafka.producer.value-serializer}")
  26.     private String valueSerializer;
  28.     public Map<String, Object> producerConfigs() {
  29.         Map<String, Object> props = new HashMap<>();
  30.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
  31.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
  32.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
  33.         props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RoundRobinPartitioner.class);
  34.         return props;
  35.     }
  37.     public ProducerFactory<String, ?> producerFactory() {
  38.         return new DefaultKafkaProducerFactory<>(producerConfigs());
  39.     }
  41.     @Bean
  42.     public KafkaTemplate<String, ?> kafkaTemplate() {
  43.         return new KafkaTemplate<>(producerFactory());
  44.     }
  47.     //第二次创建
  48.     @Bean
  49.     public NewTopic newTopic9() {
  50.         return new NewTopic("heTopic", 9, (short) 1);
  51.     }
  52. }
复制代码
2.2、application.yml文件

  1. spring:
  2.   application:
  3.     #应用名称
  4.     name: spring-boot-01-kafka-base
  6.   #kafka连接地址（ip+port）
  7.   kafka:
  8.     bootstrap-servers: <你的kafka服务器IP>:9092
  9.     #配置生产者(有24个配置)
  10.     producer:
  11.       #key默认是StringSerializer序列化
  12.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  13.       #value默认是ToStringSerializer序列化
  14.       value-serializer: org.springframework.kafka.support.serializer.ToStringSerializer
  17.     #配置消费者(有24个配置)
  18.     consumer:
  19.       auto-offset-reset: earliest
  21.     template:
  22.       default-topic: default-topic
复制代码

2.3、生产者

  1. package com.power.producer;
  3. import com.power.model.User;
  4. import org.apache.kafka.clients.producer.ProducerRecord;
  5. import org.apache.kafka.common.header.Headers;
  6. import org.apache.kafka.common.header.internals.RecordHeaders;
  7. import org.springframework.kafka.core.KafkaTemplate;
  8. import org.springframework.kafka.support.KafkaHeaders;
  9. import org.springframework.kafka.support.SendResult;
  10. import org.springframework.messaging.Message;
  11. import org.springframework.messaging.support.MessageBuilder;
  12. import org.springframework.stereotype.Component;
  13. import org.springframework.util.concurrent.ListenableFuture;
  15. import javax.annotation.Resource;
  16. import java.nio.charset.StandardCharsets;
  17. import java.util.Date;
  18. import java.util.concurrent.CompletableFuture;
  19. import java.util.concurrent.ExecutionException;
  21. @Component
  22. public class EventProducer {
  24.     @Resource
  25.     private KafkaTemplate<String,Object> kafkaTemplate2;
  27.     public void send10(){
  28.         User user = User.builder().id(1208).phone("16767667676").birthday(new Date()).build();
  29.         //分区是null，让kafka自己去决定把消息发送到哪个分区
  30.         kafkaTemplate2.send("heTopic",user);
  31.     }
  32. }
复制代码

2.4、测试类


  1. package com.power;
  3. import com.power.model.User;
  4. import com.power.producer.EventProducer;
  5. import org.junit.jupiter.api.Test;
  6. import org.springframework.boot.test.context.SpringBootTest;
  8. import javax.annotation.Resource;
  9. import java.util.Date;
  11. @SpringBootTest
  12. public class SpringBoot01KafkaBaseApplication {
  14.     @Resource
  15.     private EventProducer eventProducer;
  17.     @Test
  18.     void send10(){
  19.         for (int i = 0; i <5 ; i++) {
  20.             eventProducer.send10();
  21.         }
  23.     }
  25. }
复制代码
2.5、实行结果

实行完测试类，发现5次哀求分别发送到了kafka的heTopic主题的5个差别分区中：









3、自界说分区分配计谋


3.1、创建自界说分配计谋类

  1. package com.power.config;import org.apache.kafka.clients.producer.Partitioner;import org.apache.kafka.common.Cluster;import org.apache.kafka.common.PartitionInfo;import org.apache.kafka.common.utils.Utils;import java.util.List;import java.util.Map;import java.util.concurrent.atomic.AtomicInteger;public class CustomerPartitioner implements Partitioner {    private AtomicInteger nextPartition = new AtomicInteger(0);    @Override    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);        int numPartitions = partitions.size();        if(key==null){            //使用轮询方式选择分区            int next = nextPartition.getAndIncrement();            if(next>=numPartitions){                nextPartition.compareAndSet(next,0);            }            if(next>0){                next--;            }            System.out.println("分区值："+next);            return next;        }else {            //如果key不为inull,则使用默认的分区计谋            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
  2.         }    }    @Override    public void close() {    }    @Override    public void configure(Map<String, ?> configs) {    }}
复制代码
3.2、修改kafka配置类

指定使用自界说的分区分配类

3.3、application.yml文件

  1. spring:
  2.   application:
  3.     #应用名称
  4.     name: spring-boot-01-kafka-base
  6.   #kafka连接地址（ip+port）
  7.   kafka:
  8.     bootstrap-servers: <你的kafka服务器IP>:9092
  9.     #配置生产者(有24个配置)
  10.     producer:
  11.       #key默认是StringSerializer序列化
  12.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  13.       #value默认是ToStringSerializer序列化
  14.       value-serializer: org.springframework.kafka.support.serializer.ToStringSerializer
  17.     #配置消费者(有24个配置)
  18.     consumer:
  19.       auto-offset-reset: earliest
  21.     template:
  22.       default-topic: default-topic
复制代码

3.4、生产者

  1. package com.power.producer;
  3. import com.power.model.User;
  4. import org.apache.kafka.clients.producer.ProducerRecord;
  5. import org.apache.kafka.common.header.Headers;
  6. import org.apache.kafka.common.header.internals.RecordHeaders;
  7. import org.springframework.kafka.core.KafkaTemplate;
  8. import org.springframework.kafka.support.KafkaHeaders;
  9. import org.springframework.kafka.support.SendResult;
  10. import org.springframework.messaging.Message;
  11. import org.springframework.messaging.support.MessageBuilder;
  12. import org.springframework.stereotype.Component;
  13. import org.springframework.util.concurrent.ListenableFuture;
  15. import javax.annotation.Resource;
  16. import java.nio.charset.StandardCharsets;
  17. import java.util.Date;
  18. import java.util.concurrent.CompletableFuture;
  19. import java.util.concurrent.ExecutionException;
  21. @Component
  22. public class EventProducer {
  24.     @Resource
  25.     private KafkaTemplate<String,Object> kafkaTemplate2;
  27.     public void send10(){
  28.         User user = User.builder().id(1208).phone("16767667676").birthday(new Date()).build();
  29.         //分区是null，让kafka自己去决定把消息发送到哪个分区
  30.         kafkaTemplate2.send("heTopic",user);
  31.     }
  32. }
复制代码

3.5、测试类


  1. package com.power;
  3. import com.power.model.User;
  4. import com.power.producer.EventProducer;
  5. import org.junit.jupiter.api.Test;
  6. import org.springframework.boot.test.context.SpringBootTest;
  8. import javax.annotation.Resource;
  9. import java.util.Date;
  11. @SpringBootTest
  12. public class SpringBoot01KafkaBaseApplication {
  14.     @Resource
  15.     private EventProducer eventProducer;
  17.     @Test
  18.     void send10(){
  19.         for (int i = 0; i <5 ; i++) {
  20.             eventProducer.send10();
  21.         }
  23.     }
  25. }
复制代码
3.6、测试结果









3.7、总结

使用自界说分区计谋类尝试发送消息，发现发送的5次消息，并没有一连发送到5个挨着的分区中，查看kafka源码的org.apache.kafka.clients.producer.KafkaProducer类的doSend方法发现，每一次发送前，调用了两次计算分区的方法，导致第一个得到的分区并不会正在的发送消息。
doSend方法；
  1. private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
  2.     TopicPartition tp = null;
  3.     try {
  4.         throwIfProducerClosed();
  5.         // first make sure the metadata for the topic is available
  6.         long nowMs = time.milliseconds();
  7.         ClusterAndWaitTime clusterAndWaitTime;
  8.         try {
  9.             clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), nowMs, maxBlockTimeMs);
  10.         } catch (KafkaException e) {
  11.             if (metadata.isClosed())
  12.                 throw new KafkaException("Producer closed while send in progress", e);
  13.             throw e;
  14.         }
  15.         nowMs += clusterAndWaitTime.waitedOnMetadataMs;
  16.         long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
  17.         Cluster cluster = clusterAndWaitTime.cluster;
  18.         byte[] serializedKey;
  19.         try {
  20.             serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
  21.         } catch (ClassCastException cce) {
  22.             throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() +
  23.                     " to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() +
  24.                     " specified in key.serializer", cce);
  25.         }
  26.         byte[] serializedValue;
  27.         try {
  28.             serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());
  29.         } catch (ClassCastException cce) {
  30.             throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() +
  31.                     " to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +
  32.                     " specified in value.serializer", cce);
  33.         }
  34.         int partition = partition(record, serializedKey, serializedValue, cluster);
  35.         tp = new TopicPartition(record.topic(), partition);
  37.         setReadOnly(record.headers());
  38.         Header[] headers = record.headers().toArray();
  40.         int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(),
  41.                 compressionType, serializedKey, serializedValue, headers);
  42.         ensureValidRecordSize(serializedSize);
  43.         long timestamp = record.timestamp() == null ? nowMs : record.timestamp();
  44.         if (log.isTraceEnabled()) {
  45.             log.trace("Attempting to append record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);
  46.         }
  47.         // producer callback will make sure to call both 'callback' and interceptor callback
  48.         Callback interceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp);
  50.         if (transactionManager != null && transactionManager.isTransactional()) {
  51.             transactionManager.failIfNotReadyForSend();
  52.         }
  53.         RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
  54.                 serializedValue, headers, interceptCallback, remainingWaitMs, true, nowMs);
  56.         if (result.abortForNewBatch) {
  57.             int prevPartition = partition;
  58.             partitioner.onNewBatch(record.topic(), cluster, prevPartition);
  59.             partition = partition(record, serializedKey, serializedValue, cluster);
  60.             tp = new TopicPartition(record.topic(), partition);
  61.             if (log.isTraceEnabled()) {
  62.                 log.trace("Retrying append due to new batch creation for topic {} partition {}. The old partition was {}", record.topic(), partition, prevPartition);
  63.             }
  64.             // producer callback will make sure to call both 'callback' and interceptor callback
  65.             interceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp);
  67.             result = accumulator.append(tp, timestamp, serializedKey,
  68.                 serializedValue, headers, interceptCallback, remainingWaitMs, false, nowMs);
  69.         }
  71.         if (transactionManager != null && transactionManager.isTransactional())
  72.             transactionManager.maybeAddPartitionToTransaction(tp);
  74.         if (result.batchIsFull || result.newBatchCreated) {
  75.             log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
  76.             this.sender.wakeup();
  77.         }
  78.         return result.future;
  79.         // handling exceptions and record the errors;
  80.         // for API exceptions return them in the future,
  81.         // for other exceptions throw directly
  82.     } catch (ApiException e) {
  83.         log.debug("Exception occurred during message send:", e);
  84.         if (callback != null)
  85.             callback.onCompletion(null, e);
  86.         this.errors.record();
  87.         this.interceptors.onSendError(record, tp, e);
  88.         return new FutureFailure(e);
  89.     } catch (InterruptedException e) {
  90.         this.errors.record();
  91.         this.interceptors.onSendError(record, tp, e);
  92.         throw new InterruptException(e);
  93.     } catch (KafkaException e) {
  94.         this.errors.record();
  95.         this.interceptors.onSendError(record, tp, e);
  96.         throw e;
  97.     } catch (Exception e) {
  98.         // we notify interceptor about all exceptions, since onSend is called before anything else in this method
  99.         this.interceptors.onSendError(record, tp, e);
  100.         throw e;
  101.     }
  102. }
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

乌市泽哥

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表