kafka生产者专题（原理+拦截器+序列化+分区+数据可靠+数据去重+事务） ...

熊熊出没  金牌会员 | 2025-1-8 08:55:28 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 888|帖子 888|积分 2664

生产者

发送数据原理


说明

  • 拦截器允许有多个，可以构成拦截器链
  • 生产者发送的消息会被分配到差别的分区（Partition）。每个分区在内存中都有一个对应的缓冲区（RecordAccumulator），用于暂存即将发送的消息。sender线程从中读取数据
  • sender线程两个紧张参数（巨细默认16KB，当数据量达到16KB时读取，读取时间默认0ms，当达到读取时间时，自动读取数据，不管巨细有没有达到），这两个参数均可以调整
  • NetworkClient负责将生产者的请求（如发送消息、获取元数据等）发送到相应的broker，并存储这些请求的响应；NetworkClient还负责处理网络连接的创建、维护和关闭。
  • 当生产者发送消息到broker时，可以选择差别的应答级别（acks参数）：
    acks=0：生产者不等待broker的应答，直接以为消息发送成功。这种方式性能最高，但可靠性最低。
    acks=1：生产者等待leader broker的应答，只要leader broker确认收到消息，就以为消息发送成功。这种方式性能较高，但可靠性略低。
    acks=all（或acks=-1）：leader和follower都落地回应，才以为消息发送成功。这种方式性能最低，但可靠性最高。
    Broker在收到消息后，会根据配置的应答机制向生产者发送应答（或错误）信息。
  • 请注意关于broker的落地是指数据存储到磁盘或长期化
  • 数据发送成功后（接收到应答）删除缓冲区内对应的数据
参数说明

参数默认值作用形貌bootstrap.serversnode2:9092[,node3:9092][,node4:9092]生产者连接集群所需的broker地址清单，一个或多个（逗号隔开）key.serializer（无）指定发送消息的key的序列化类型，必须写全类名value.serializer（无）指定发送消息的value的序列化类型，必须写全类名buffer.memory32MRecordAccumulator缓冲区总巨细batch.size16K缓冲区一批数据最大值，适当增加可进步吞吐量，但可能增加延迟linger.ms0ms（表示没有延迟）如果数据未达到batch.size，sender等待linger.time后发送数据acks-1应答机制：0-不需要应答，1-Leader应答，-1（all）-全部节点应答max.in.flight.requests.per.connection5允许最多没有返回ack的次数，开启幂等性时建议1-5之间enable.idempotencetrue是否开启幂等性，默认开启retries2147483647（int最大值）消息发送错误时的重试次数retry.backoff.ms100ms两次重试之间的时间间隔compression.typenone生产者发送的全部数据的压缩方式，默认不压缩 代码示例（同步发送数据）

  1. package com.wunaiieq;
  4. import org.apache.kafka.clients.producer.KafkaProducer;
  5. import org.apache.kafka.clients.producer.ProducerConfig;
  6. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import org.apache.kafka.common.serialization.StringSerializer;
  8. import java.util.Properties;
  9. import java.util.concurrent.ExecutionException;
  12. public class SyncCustomProducer {
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         //TODO 1.声明并实例化Kafka Producer的配置文件对象
  15.         Properties prop = new Properties();
  16.         //TODO 2.为配置文件对象设置参数
  17.         //TODO 2.1 配置bootstrap_server(生产者连接集群所需的broker地址清单)
  18.         prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  19.                 "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
  20.         //TODO 2.2 配置key和value的序列化类
  21.         // 设置序列化器：指定key和value的序列化类为StringSerializer，用于将字符串类型的key和value转换为字节数组，以便发送到Kafka。
  22.         prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  23.         prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  24.         //TODO 3.声明并实例化生产者对象
  25.         KafkaProducer<String,String> producer =
  26.                 new KafkaProducer<String, String>(prop);
  27.         //TODO 4.发送消息
  28.         // producer.send(...).get()：同步发送消息，send()方法返回一个Future对象，调用get()方法等待发送完成并获取结果。
  29.         for(int i = 0;i<5;i++){
  30.             //同步发送消息
  31.             producer.send(new ProducerRecord<>("topicA","sync_msg"+i)).get();
  32.         }
  33.         //TODO 5.关闭生产者
  34.         producer.close();
  35.     }
  36. }
复制代码
代码说明
类/对象形貌用途PropertiesJava标准库中的类，用于维护键值对列表。 Properties类提供了一种方便的方式来读取和写入属性文件（通常是.properties文件）在本代码中用于存储Kafka生产者的配置参数。KafkaProducer<K, V>Kafka客户端库中的类，用于向Kafka主题发送消息。泛型参数K和V分别表示消息键和值的类型。创建生产者实例，发送消息到Kafka主题。ProducerConfigKafka客户端库中的类，包含生产者配置的常量。提供配置参数的常量值，如broker地址、序列化器等。ProducerRecord<K, V>Kafka客户端库中的类，表示要发送到Kafka主题的消息记录。泛型参数K和V分别表示消息键和值的类型。创建消息记录对象，包含主题、键和值。StringSerializerKafka客户端库中的类，实现了Serializer<String>接口，用于将字符串类型的键或值序列化为字节数组。作为键和值的序列化器，将字符串转换为字节数组进行传输。 效果

代码示例（异步）

  1. package com.wunaiieq;
  3. import org.apache.kafka.clients.producer.KafkaProducer;
  4. import org.apache.kafka.clients.producer.ProducerConfig;
  5. import org.apache.kafka.clients.producer.ProducerRecord;
  6. import org.apache.kafka.common.serialization.StringSerializer;
  8. import java.util.Properties;
  10. public class UnSyncCustomProducer {
  11.     public static void main(String[] args) {
  12.         //实例化Properties
  13.         Properties prop = new Properties();
  14.         //集群节点
  15.         prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  16.                 "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
  17.         //key和value
  18.         prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  19.         prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  20.         //创建kafka生产者对象，并写入响应参数
  21.         KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);
  22.         //发送数据
  23.         for (int i = 0; i < 5; i++) {
  24.             //异步发送数据，不调用get方法
  25.             producer.send(new ProducerRecord<>(
  26.                     "topicA", "unsync_msg" + i
  27.             ));
  28.         }
  29.         producer.close();
  30.     }
  31. }
复制代码
异步和同步的区别

同步发送

界说与流程

界说： 同步发送是指生产者在发送一条消息后，会立即等待Kafka服务器的响应。只有在服务器返回成功响应后，生产者才会继续发送下一条消息。
流程：
生产者调用send()方法发送消息。
send()方法返回一个Future对象。
生产者调用Future对象的get()方法，该方法会阻塞当前线程，直到Kafka服务器返反响应或抛出异常。
生产者收到响应后，根据结果（成功或失败）进行后续操作。
特点

高可靠性：同步发送确保每条消息都被Kafka集群接收并长期化。生产者等待Kafka确认消息已经成功写入指定分区且复制到满足副本因子的节点上，从而进步了消息的可靠性。
异常处理：如果发送过程中发生异常，生产者可以立即感知并处理，避免了消息的丢失。
性能较低：由于同步发送需要阻塞等待响应，因此会增加消息的延迟，降低系统的吞吐量。特别是在高并发场景下，可能会导致线程资源的大量占用和性能瓶颈。
易调试：便于发现和处理异常，有利于开发和测试阶段的调试工作。
异步发送

界说与流程

界说： 异步发送是指生产者在发送一条消息后，不会立即等待Kafka服务器的响应，而是继续发送下一条消息。发送方通过通报一个回调函数给send()方法，该回调函数将在消息发送结果（成功或失败）可用时被异步调用。
流程：
生产者调用send()方法发送消息，并通报一个回调函数。
Kafka客户端将消息放入内部缓冲区，并立即返回。
Sender线程负责将缓冲区中的消息批量发送到Kafka集群。
当消息发送成功或失败时，Kafka客户端调用之前通报的回调函数，通知生产者消息发送的结果。
特点

高性能：异步发送方式下，生产者无需等待每个消息的确认即可继续发送下一条消息，从而进步了消息的发送效率，适用于高吞吐量场景。
灵活性：通过回调函数，生产者可以对消息发送的结果进行异步处理，如记录日记、重试发送等。
可靠性相对较低：由于生产者不会立即得知消息是否成功写入Kafka，因此消息的可靠性需要额外关注。如果生产者在发送消息后立即崩溃，可能会导致部分消息丢失。
调试复杂：由于消息发送和结果是异步的，因此调试时可能需要更多的日记记录和监控手段来确保消息的可靠性和完整性。
异步回调

形貌

回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。消息发送失败后会自动重试，不需要再回调函数中手动重试。
代码示例

  1. package com.wunaiieq;
  2. import org.apache.kafka.clients.producer.*;
  3. import org.apache.kafka.common.serialization.StringSerializer;
  4. import java.util.Properties;
  5. import java.util.concurrent.ExecutionException;
  6. public class UnSyncCallBackCustomProducer {
  7.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  8.         //TODO 1.声明并实例化Kafka Producer的配置文件对象
  9.         Properties prop = new Properties();
  10.         //TODO 2.为配置文件对象设置参数
  11.         // 2.1 配置bootstrap_servers
  12.         prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  13.                 "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
  14.         // 2.2 配置key和value的序列化类
  15.         prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  16.         prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  17.         //TODO 3.声明并实例化生产者对象
  18.         KafkaProducer<String,String> producer = new KafkaProducer<String, String>(prop);
  19.         //TODO 4.发送消息
  20.         for(int i = 0;i<5;i++){
  21.             //异步发送消息 不调用get()方法
  22.             producer.send(new ProducerRecord<>("topicA", "unsync_msg" + i),
  23.                     new Callback() {
  24.                         //如下方法在生产者收到acks确认时异步调用
  25.                         @Override
  26.                         public void onCompletion(RecordMetadata recordMetadata, Exception e) {
  27.                             if(e == null){
  28.                                 //无异常信息,笑死发送成功,输出主题和分区信息到控制台
  29.                                 System.out.println("topic:"+recordMetadata.topic()
  30.                                         +",partition:"+recordMetadata.partition());
  31.                             }else{//打印异常信息
  32.                                 System.out.println(e.getMessage());
  33.                             }
  34.                         }
  35.                     });
  36.             Thread.sleep(5);
  37.         }
  38.         //TODO 5.关闭生产者
  39.         producer.close();
  40.     }
  41. }
复制代码
拦截器

形貌

拦截器(Interceptor)是kafka0.10.0.0版本中引入的新功能，主要用于实现clients端的定制化控制逻辑。它可以使得用户在消息发送前以及producer回调逻辑前有机会对消息做一些定制化需求，好比修改消息等。同时允许指定多个Interceptor按序作用于同一条消息从而形成一个拦截器链（Interceptor Chain）。
自界说拦截器需要实现org.apache.kafka.clients.producer.ProducerInterceptor接口。
拦截器内部方法

  • onSend 方法

    • 作用：在消息发送之进步行拦截，允许对消息进行修改或处理。
    • 参数：接收一个ProducerRecord对象。
    • 返回值：返回一个ProducerRecord对象，可能是修改后的记录。
    • 应用场景：添加消息头、修改消息内容、过滤消息等。

  • onAcknowledgement 方法

    • 作用：在消息发送成功或失败后进行回调。
    • 参数：

      • RecordMetadata：包含消息的元数据。
      • Exception：发送过程中可能抛出的异常，成功时为null。

    • 返回值：无。
    • 应用场景：记录发送结果、统计发送成功率、处剃头送失败等。

  • close 方法

    • 作用：在拦截器不再使用时进行资源清理。
    • 参数：无。
    • 返回值：无。
    • 应用场景：关闭打开的文件、释放内存、断开网络连接等。

拦截器Interceptor可能运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。别的，若指定了多个Interceptor,则producer将按照指定顺序调用它们，同时把每个Interceptor中捕获的异常记录写入到错误日记中而不是向上通报。这在使用过程中要特别注意。
代码示例

实现一个简单的双interceptor构成的拦截链。
第一个Interceptor会在消息发送前将时间戳信息加到消息value的前面
第二个Interceptor在消息发送后更新成功发送消息数或失败发送消息数。
第一个拦截器示例
  1. package com.wunaiieq;
  4. import org.apache.kafka.clients.producer.ProducerInterceptor;
  5. import org.apache.kafka.clients.producer.ProducerRecord;
  6. import org.apache.kafka.clients.producer.RecordMetadata;
  9. import java.util.Map;
  12. public class TimeStampInterceptor implements ProducerInterceptor<String,String> {
  13.     /**初始化拦截器，并接收Kafka生产者的配置参数。
  14.      * */
  15.     @Override
  16.     public void configure(Map<String, ?> configs) {
  17.     }
  18.     /**发送之前被调用，对消息进行处理。
  19.      * */
  20.     @Override
  21.     public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
  22.         return new ProducerRecord<String, String>(
  23.                 //原始消息记录的主题、分区、时间戳和键。
  24.                 record.topic(),record.partition(),
  25.                 record.timestamp(), record.key(),
  26.                 //将当前系统时间戳（System.currentTimeMillis()）和原始消息值拼接成新的消息值，中间用逗号分隔。
  27.                 System.currentTimeMillis()+","+record.value());
  28.     }
  29.     /**消息发送成功或失败后被调用。
  30.      * */
  31.     @Override
  32.     public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
  33.     }
  34.     /**在拦截器不再使用时进行资源清理。
  35.      * */
  36.     @Override
  37.     public void close() {
  38.     }
  39. }
复制代码
第二个拦截器示例
  1. package com.wunaiieq;
  4. import org.apache.kafka.clients.producer.ProducerInterceptor;
  5. import org.apache.kafka.clients.producer.ProducerRecord;
  6. import org.apache.kafka.clients.producer.RecordMetadata;
  9. import java.util.Map;
  12. public class CounterIntercepter implements ProducerInterceptor<String,String> {
  13.     private int errorCounter = 0;
  14.     private int successCounter = 0;
  15.     /**onSend方法，该方法在消息发送之前被调用，用于对消息进行处理。
  16.      * 由于这是第二个拦截器，因此这里接受的是前一个拦截器的输出
  17.      * */
  18.     @Override
  19.     public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
  20.         return record;
  21.     }
  22.     /**在消息发送成功或失败后被调用。
  23.      * 统计消息发送成功或失败的数量
  24.      * */
  25.     @Override
  26.     public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
  27.         if(exception==null){
  28.             successCounter++;
  29.         }else{
  30.             errorCounter++;
  31.         }
  32.     }
  33.     /**拦截器关闭时会进行的额外操作
  34.      * 打印成功或失败的消息数量
  35.      * */
  36.     @Override
  37.     public void close() {
  38.         System.out.println("successful send:"+successCounter);
  39.         System.out.println("failed send:"+errorCounter);
  40.     }
  41.     @Override
  42.     public void configure(Map<String, ?> configs) {
  43.     }
  44. }
复制代码
拦截器调用
  1. package com.wunaiieq;
  4. import org.apache.kafka.clients.producer.KafkaProducer;
  5. import org.apache.kafka.clients.producer.ProducerConfig;
  6. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import org.apache.kafka.common.serialization.StringSerializer;
  10. import java.util.ArrayList;
  11. import java.util.List;
  12. import java.util.Properties;
  13. import java.util.concurrent.ExecutionException;
  16. public class SyncCustomProducerInterceptor {
  17.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  18.         Properties prop = new Properties();
  19.         prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  20.                 "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
  21.         prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  22.         prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  23.         // 构造拦截器链
  24.         List<String> interceptors = new ArrayList<>();
  25.         interceptors.add("com.wunaiieq.TimeStampInterceptor");
  26.         interceptors.add("com.wunaiieq.CounterIntercepter");
  27.         //配置拦截器链(将拦截器链加入到配置文件中)
  28.         prop.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
  30.         KafkaProducer<String, String> producer =
  31.                 new KafkaProducer<String, String>(prop);
  32.         for (int i = 5; i < 10; i++) {
  33.             //同步发送消息
  34.             producer.send(new ProducerRecord<>("topicA", "sync_msg" + i)).get();
  35.         }
  36.         //一定要关闭生产者，这样才会调用interceptor的close方法
  37.         producer.close();
  38.     }
  39. }
复制代码
效果


消息序列化

形貌


消息序列化是将对象转换为字节流的过程。在Kafka中，生产者需要将消息对象序列化为字节流，以便通过网络发送给Kafka集群；而消费者则需要将从Kafka集群接收到的字节流反序列化为对象，以便进行后续处理。
代码示例（自界说序列化）

pom.xml
增加依靠
  1.     <dependency>
  2.       <groupId>org.codehaus.jackson</groupId>
  3.       <artifactId>jackson-mapper-asl</artifactId>
  4.       <version>1.9.13</version>
  5.     </dependency>
复制代码
UserVo.java
值对象
  1. package com.wunaiieq;
  4. public class UserVo {
  5.     private String name;
  6.     private int age;
  7.     private String address;
  8.     public UserVo(String name, int age, String address) {
  9.         this.name = name;
  10.         this.age = age;
  11.         this.address = address;
  12.     }
  13.     @Override
  14.     public String toString() {
  15.         return "UserVo{" +
  16.                 "name='" + name + '\'' +
  17.                 ", age=" + age +
  18.                 ", address='" + address + '\'' +
  19.                 '}';
  20.     }
  21.     public String getName() {
  22.         return name;
  23.     }
  26.     public void setName(String name) {
  27.         this.name = name;
  28.     }
  31.     public int getAge() {
  32.         return age;
  33.     }
  36.     public void setAge(int age) {
  37.         this.age = age;
  38.     }
  41.     public String getAddress() {
  42.         return address;
  43.     }
  46.     public void setAddress(String address) {
  47.         this.address = address;
  48.     }
  49. }
复制代码
UserSerializer.java
重写Serializer接口实现序列化操作
  1. package com.wunaiieq;
  4. import org.apache.kafka.common.errors.SerializationException;
  5. import org.apache.kafka.common.serialization.Serializer;
  6. import org.codehaus.jackson.map.ObjectMapper;
  7. import java.io.IOException;
  8. import java.nio.charset.StandardCharsets;
  9. import java.util.Map;
  12. public class UserSerializer implements Serializer<UserVo> {
  13.     private ObjectMapper objectMapper;
  14.     @Override
  15.     public void configure(Map<String, ?> configs, boolean isKey) {
  16.         objectMapper = new ObjectMapper();
  17.         //Serializer.super.configure(configs, isKey);
  18.     }
  20.     /**
  21.      * @param topic 消息将要发送到的主题名
  22.      * @param data 需要序列化的UserVo对象。
  23.      * */
  24.     @Override
  25.     public byte[] serialize(String topic, UserVo data) {
  26.         //存储序列化后的字节数组
  27.         byte[] ret = null;
  28.         try {
  29.             //data写成JSON字符串再写成UTF_8的字节数组
  30.             ret = objectMapper.writeValueAsString(data)
  31.                     .getBytes(StandardCharsets.UTF_8);
  32.         } catch (IOException e) {
  33.             throw new SerializationException("Error when serializing UserVo to byte[],exception is " + e.getMessage());
  34.         }
  35.         return ret;
  36.     }
  39.     @Override
  40.     public void close() {
  41.         objectMapper = null;
  42.         //Serializer.super.close();
  43.     }
  44. }
复制代码
UserSerProducer.java
调用自界说序列化机制
  1. package com.wunaiieq;
  4. import org.apache.kafka.clients.producer.*;
  5. import org.apache.kafka.common.serialization.StringSerializer;
  8. import java.util.Properties;
  9. import java.util.concurrent.ExecutionException;
  12. public class UserSerProducer {
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         Properties prop = new Properties();
  15.         prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  16.                 "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
  17.         // TODO 不用修改key的序列化机制，后续没用到
  18.         prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  19.         // TODO 修改value的序列化机制
  20.         prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserSerializer.class.getName());
  21.         KafkaProducer<String,UserVo> producer = new KafkaProducer<String, UserVo>(prop);
  22.         UserVo userVo = new UserVo("wunaiieq",18,"北京");
  23.         producer.send(
  24.                 // TODO 关于消息记录的构造中，可以指定 1.主题、值 2.主题、键、值
  25.                 new ProducerRecord<String,UserVo>("topicA", userVo),
  26.                 new Callback() {
  27.                     //如下方法在生产者收到acks确认时异步调用
  28.                     @Override
  29.                     public void onCompletion(RecordMetadata recordMetadata, Exception e) {
  30.                         if(e == null){
  31.                             //无异常信息，输出主题和分区信息到控制台
  32.                             System.out.println("topic:"+recordMetadata.topic()
  33.                                     +",partition:"+recordMetadata.partition());
  34.                         }else{//打印异常信息
  35.                             System.out.println(e.getMessage());
  36.                         }
  37.                     }
  38.                 });
  39.         Thread.sleep(50);
  40.         producer.close();
  41.     }
  42. }
复制代码
效果
前面的不消管，只是没清空而已


分区

形貌

分区位于拦截器链后面

生产者分区的优势：


  • 便于公道使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。公道控制分区的任务，可以实现负载平衡的效果。
  • 进步并行度，生产者可以以分区为单元发送数据；消费者可以以分区为单元进行消费数据。
  • 分区后，更方便于做副本备份，进步了数据安全性。
分区计谋

以下为提供的默认分区计谋，自行选择即可

  • 轮询计谋（Round-Robin Strategy）
    原理：按照顺序将消息发送到差别的分区，每个消息被发送到其对应分区，循环轮询每个分区，确保消息在全部分区之间均匀分布。
    特点：适用于生产者不需要根据消息内容或键选择特定分区的场景，能够实现负载平衡，最大限度地使用集群资源。
    默认情况：这是Kafka Java生产者API默认提供的分区计谋。如果没有指定分区计谋，则会默认使用轮询计谋。
  • 按键分配计谋（Key-Based Partitioning）
    原理：消息的键被用作决定消息分区的依据。生产者将消息的键发送给Kafka，Kafka根据键的哈希值将消息路由到相应的分区。
    特点：适用于键值对的数据布局，通过将具有雷同键的消息发送到同一分区，可以进步数据局部性和处理效率。同时，能够包管具有雷同键的消息顺序性。
  • 范围分区计谋（Range Partitioning）
    原理：根据消息键的范围将消息分配到差别的分区。每个分区包含一个键值范围内的消息。
    特点：适用于有序数据的处理，如时间戳或递增的ID。通过将具有相似时间戳或递增ID的消息分配到同一分区，可以进步处理效率并包管数据的顺序性。
  • 自界说分区计谋（Custom Partitioning）
    原理：用户可以根据特定的业务逻辑或规则来决定消息的分区。通过实现自界说的分区器类，根据应用步伐的需求来界说分区的逻辑。
    特点：提供了更高的灵活性，可以根据地理位置、用户ID或其他业务规则来决定消息的分区。
    实现方式：实现org.apache.kafka.clients.producer.Partitioner接口，并重写partition、close和configure方法。其中，partition方法是核心，用于根据给定的键、值和分区信息来计算分区号。
  • 粘性分区计谋（Sticky Partitioning）
    原理：尽可能将消息分配到与之前消息雷同的分区，以减少跨分区的数据移动和复制。通过维护一个分区和消费者的映射关系来实现。
    特点：在消费者组或分区数目发生变化时，能够尽可能减少对现有分区分配的影响，减少负载平衡的开销，进步处理效率。
代码示例

写入默认分区（0号分区）

消息只会发送到指定的分区内部
  1. package com.wunaiieq.partition;
  4. import org.apache.kafka.clients.producer.*;
  5. import org.apache.kafka.common.serialization.StringSerializer;
  8. import java.util.Properties;
  9. import java.util.concurrent.ExecutionException;
  12. public class ProducerToPartition {
  13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  14.         //TODO 1.声明并实例化Kafka Producer的配置文件对象
  15.         Properties prop = new Properties();
  16.         //TODO 2.为配置文件对象设置参数
  17.         // 2.1 配置bootstrap_servers
  18.         prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  19.                 "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
  20.         // 2.2 配置key和value的序列化类
  21.         prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  22.         prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  23.         //TODO 3.声明并实例化生产者对象
  24.         KafkaProducer<String,String> producer = new KafkaProducer<String, String>(prop);
  25.         //TODO 4.发送消息
  26.         for(int i = 0;i<5;i++){
  27.             //指定数据发送到0号分区，key为null
  28.             producer.send(new ProducerRecord<>("topicA",0,null, "unsync_msg" + i),
  29.                     new Callback() {
  30.                         //如下方法在生产者收到acks确认时异步调用
  31.                         @Override
  32.                         public void onCompletion(RecordMetadata recordMetadata, Exception e) {
  33.                             if(e == null){
  34.                                 //无异常信息，输出主题和分区信息到控制台
  35.                                 System.out.println("topic:"+recordMetadata.topic()
  36.                                         +",partition:"+recordMetadata.partition());
  37.                             }else{//打印异常信息
  38.                                 System.out.println(e.getMessage());
  39.                             }
  40.                         }
  41.                     });
  42.             Thread.sleep(5);
  43.         }
  44.         //TODO 5.关闭生产者
  45.         producer.close();
  46.     }
  47. }
复制代码
自界说分区机制

部分消息可能需要额外的处理内容，好比审计等等，这类消息的key会携带关键字符串“wunaiieq”，如今让其发送到topicA主题的末了一个分区上，以便于后续处理，其他的消息则随机发送（不包罗末了一个分区）
WunaiieqPartitioner.java
分区写入计谋
  1. package com.wunaiieq.partition;
  4. import org.apache.kafka.clients.producer.Partitioner;
  5. import org.apache.kafka.common.Cluster;
  6. import org.apache.kafka.common.PartitionInfo;
  9. import java.util.List;
  10. import java.util.Map;
  11. import java.util.Random;
  14. public class WunaiieqPartitioner implements Partitioner {
  15.     private Random random;
  16.     @Override
  17.     public void configure(Map<String, ?> configs) {
  18.         //该方法实现必要资源的初始化工作
  19.         random = new Random();
  20.     }
  21.     /** 计算信息对应的分区
  22.      * @param topic   主题
  23.      * @param key    消息的key
  24.      * @param keyBytes  消息的key序列化后的字节数组
  25.      * @param value   消息的value
  26.      * @param valueBytes 消息value序列化后的字节数组
  27.      * @param cluster  集群元数据 可以获取分区信息
  28.      * @return  息对应的分区号
  29.      */
  30.     @Override
  31.     public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
  32.         //将key转换为字符串
  33.         String keyInfo = (String)key;
  34.         //获取主题的分区对象列表
  35.         List<PartitionInfo> partitionInfoList = cluster.availablePartitionsForTopic(topic);
  36.         //获取主题下的分区总数量
  37.         int partCount = partitionInfoList.size();
  38.         if (partCount <= 1) {
  39.             System.out.println("1 partition");
  40.             return 0;  // 只有一个分区时，直接返回0
  41.         }
  42.         //最后一个分区号
  43.         int wunaiieqPartition = partCount-1;
  44.         //如果 key 为空、key 为空字符串或 key 不包含 "wunaiieq"，则随机选择一个除最后一个分区外的分区；否则，消息发送到最后一个分区。
  45.         return keyInfo==null || keyInfo.isEmpty()
  46.                 ||!keyInfo.contains("wunaiieq")
  47.                 ? random.nextInt(partCount-1) : wunaiieqPartition ;
  48.     }
  49.     @Override
  50.     public void close() {
  51.         //该方法实现必要资源的清理工作
  52.         random = null;
  53.     }
  54. }
复制代码
CustomPartitionerProducer.java
调用分区计谋
  1. package com.wunaiieq.partition;
  4. import org.apache.kafka.clients.producer.*;
  5. import org.apache.kafka.common.serialization.StringSerializer;
  6. import java.util.Properties;
  7. import java.util.concurrent.ExecutionException;
  10. public class CustomPartitionerProducer {
  11.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  12.         Properties prop = new Properties();
  13.         prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  14.                 "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
  15.         prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  16.         prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  17.         // TODO 在用于构造KafkaProducer的Properties对象中设置partitioner.class参数
  18.         prop.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,
  19.                 "com.wunaiieq.partition.WunaiieqPartitioner");
  20.         KafkaProducer<String,String> producer = new KafkaProducer<String, String>(prop);
  21.         for(int i = 0;i<5;i++){
  22.             // TODO 不指定分区号，key为"wunaiieq"测试运行一次，改为"kafka"后再测试一次。
  23.             producer.send(new ProducerRecord<>("topicA","aa", "unsync_msg" + i),
  24.                     new Callback() {
  25.                         //如下方法在生产者收到acks确认时异步调用
  26.                         @Override
  27.                         public void onCompletion(RecordMetadata recordMetadata, Exception e) {
  28.                             if(e == null){
  29.                                 //无异常信息，输出主题和分区信息到控制台
  30.                                 System.out.println("topic:"+recordMetadata.topic()
  31.                                         +",partition:"+recordMetadata.partition());
  32.                             }else{//打印异常信息
  33.                                 System.out.println(e.getMessage());
  34.                             }
  35.                         }
  36.                     });
  37.             Thread.sleep(5);
  38.         }
  39.         producer.close();
  40.     }
  41. }
复制代码
消息丢失


判断消息丢失时，一般看应答机制

  • acks=0：由于不需要等待leader数据长期化就完成应答，leader宕机后可能存在数据丢失（follower内部数据从leader中同步，leader没有---->>follower也没有）
  • acks=1：此时leader长期化完成应答（但是follower可能没有完成数据同步，leader宕机，导致数据丢失）一般用于传输普通日记
  • acks=all（或acks=-1）：leader和follower都长期化后并回应，才以为消息发送成功。这种方式性能最低，但可靠性最高。传输紧张数据。
特别情况
在acks=-1或all的情况下，Leader接收到数据并长期化后，全部Follower开始同步Leader刚刚长期化的数据，但是有一个Follower因故障迟迟不能进行数据同步，该问题应该怎么解决？

Leader维护了一个动态的in-sync replica set（ISR），意为和Leader保持同步的Follower+Leader聚集(leader：0，isr:0,1,2)。
如果Follower长时间未向Leader发送通讯请求或同步数据，则该Follower将被踢出ISR。
该时间阈值由replica.lag.time.max.ms参数设定，默认30000ms。比方1超时，(leader:0, isr:0,2)。如许就不消等长期联系不上或者已经故障的节点。
消息绝对不丢失的条件



  • ACK级别设置为-1
  • 分区副本>=2
  • ISR应答的最小副本数>=2 （最小副本数有min.insync.replicas设置，默以为1）
生产中配置响应级别代码块：
  1. // 设置 acks
  2. prop.put(ProducerConfig.ACKS_CONFIG, "all");
  3. //重试次数 retries，默认是 int 最大值，2147483647
  4. prop.put(ProducerConfig.RETRIES_CONFIG, 3);
复制代码
数据去重

形貌

数据重复的缘故原由


  • 当生产者发送消息到Kafka集群时，如果由于网络故障或Kafka Broker的临时问题导致消息发送失败，生产者通常会进行重试。如果重试时Kafka Broker已经成功处理了之前的消息但尚未发送确认（ACK），那么重试发送的消息就会导致数据重复。
  • 网络延迟或不稳固可能导致消息发送失败，生产者会进行重试，从而增加消息重复的风险。
  • 如果Kafka Broker在消息发送成功后崩溃，但在发送确认（ACK）之前崩溃，生产者可能会重试发送雷同的消息，导致消息重复。
数据去重


  • 至少一次（At Least Once）：ACK级别设置为-1 + 分区副本大于即是2 + ISR里应答的最小副本数目大于即是2（包管数据绝对不丢失，但不能包管数据不重复）
  • 最多一次（At Most Once）：ACK级别设置为0（包管数据绝对不重复，但不能包管数据不丢失）
  • 准确一次（Exactly Once）：对于一些非常紧张的信息，好比和钱相干的数据，要求数据既不能重复也不丢失。
幂等性

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会长期化一条，包管了不重复。


  • 准确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数目>=2） 。
  • 重复数据的判断标准：具有<ID, Partition, SeqNumber>雷同主键的消息提交时，Broker只会长期化一条。其中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。所以幂等性只能包管的是在单分区单会话内不重复。

使用幂等性的使用
开启参数 enable.idempotence 默以为 true，false 关闭。
事务


   

  • 寻找事务协调器：KafkaProducer 使用 trans.id 寻找事务协调器 (Transaction Coordinator)。
  • 协调器通过 broker0 返回事务协调器的地址，包罗事务信息的主题和分区领导者 (transaction_state-分区-Leader)。
  • 请求 partId（开启幂等性）： KafkaProducer向事务协调器请求 partId，并开启幂等性。
  • 事务协调器接收到请求后，将请求长期化，并返回 partId给 KafkaProducer。
  • 发送消息： KafkaProducer 使用返回的 partId 发送消息到指定主题的指定分区(topicA-Partition0 或 topicA-Partition1)。
    6.发送 commit 请求： KafkaProducer发送 commit 请求到事务协调器，以提交事务。
    7.事务协调器接收到 commit 请求后，将其长期化。
  • 事务成功：事务协调器确认事务成功，并返回成功信息给 KafkaProducer。
  代码示例（事务）

  1. package com.wunaiieq;
  2. import org.apache.kafka.clients.producer.KafkaProducer;
  3. import org.apache.kafka.clients.producer.ProducerConfig;
  4. import org.apache.kafka.clients.producer.ProducerRecord;
  5. import org.apache.kafka.common.serialization.StringSerializer;
  6. import java.util.Properties;
  7. import java.util.concurrent.ExecutionException;
  8. public class ProducerTransaction {
  9.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  10.         Properties prop = new Properties();
  11.         prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  12.                 "192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");
  13.         prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  14.         prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  15.         //TODO 设置事务id
  16.         prop.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
  17.                 "transaction_id_topicA_0");
  18.         KafkaProducer<String,String> producer = new KafkaProducer<String, String>(prop);
  19.         //TODO 初始化事务
  20.         producer.initTransactions();
  21.         //TODO 开启事务
  22.         producer.beginTransaction();
  23.         //TODO 添加异常处理，成功提交事务，失败回滚事务
  24.         try {
  25.             //发送消息
  26.             for (int i = 0; i < 5; i++) {
  27.                 //同步发送消息
  28.                 producer.send(new ProducerRecord<>("topicA", "sync_msg" + i)).get();
  29.             }
  30.             //TODO 提交事务
  31.             producer.commitTransaction();
  32.         }catch (Exception e){
  33.             //TODO 放弃事务
  34.             producer.abortTransaction();
  35.         }
  36.         producer.close();
  37.     }
  38. }
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

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

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

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

本版积分规则

熊熊出没

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

楼主热帖

标签云

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