一文弄懂如何创建并使用Kafka生产者

Posted 黑马程序员官方

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了一文弄懂如何创建并使用Kafka生产者相关的知识,希望对你有一定的参考价值。

今天文章内容大家看完后可以:

  • 深入学习Kafka数据产生大致流程
  • 如何创建并使用Kafka生产者
  • Kafka生产者常用配置

一、消息发送

1.1 Kafka Java客户端数据生产流程解析

  • ①、首先要构造一个 ProducerRecord 对象,该对象可以声明主题Topic、分区Partition、键 Key以及值 Value,主题和值是必须要声明的,分区和键可以不用指定。
  • ②、调用send() 方法进行消息发送。
  • ③、因为消息要到网络上进行传输,所以必须进行序列化,序列化器的作用就是把消息的 key 和value对象序列化成字节数组。 后,生产者就知道该往哪个主题和分区发送记录了。
  • ④、接着这条记录会被添加到一个记录批次里面,这个批次里所有的消息会被发送到相同的主题和分区。会有一个独立的线程来把这些记录批次发送到相应的 Broker 上。
  • ⑤、Broker成功接收到消息,表示发送成功,返回消息的元数据(包括主题和分区信息以及记录在分区里的偏移量)。发送失败,可以选择重试或者直接抛出异常。

依赖的包 <kafka.version>2.0.0</kafka.version>

1.2 必要参数的配置

见代码库:com.heima.kafka.chapter2.KafkaProducerAnalysis

public static Properties initConfig() 
        Properties props = new Properties();
        // 该属性指定 brokers 的地址清单,格式为 host:port。清单里不需要包含所有的 broker地址,
        // 生产者会从给定的 broker 里查找到其它 broker 的信息。——建议至少提供两个 broker的信息,因为一旦其中一个宕机,生产者仍然能够连接到集群上。
props.put("bootstrap.servers", brokerList);
        // 将 key 转换为字节数组的配置,必须设定为一个实现了
org.apache.kafka.common.serialization.Serializer 接口的类,
        // 生产者会用这个类把键对象序列化为字节数组。
        // ——kafka 默认提供了 StringSerializer和 IntegerSerializer、
ByteArraySerializer。当然也可以自定义序列化器。
        props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
        // 和 key.serializer 一样,用于 value 的序列化
        props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
        // 内容形式如:"producer-1"
        props.put("client.id", "producer.client.id.demo");
        return props;
    

Properties props = initConfig(); 
        KafkaProducer<String, String> producer = new KafkaProducer<>(props); 
// KafkaProducer<String, String> producer = new KafkaProducer<>(props, 
//         new StringSerializer(), new StringSerializer()); 
        //生成 ProducerRecord 对象,并制定 Topic,key 以及 value 
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, 
"hello, Kafka!"); 
        try  
            // 发送消息 
            producer.send(record); 

1.3 发送类型

发送即忘记

producer.send(record)

同步发送

//通过send()发送完消息后返回一个Future对象,然后调用Future对象的get()方法等待kafka响应
//如果kafka正常响应,返回一个RecordMetadata对象,该对象存储消息的偏移量
// 如果kafka发生错误,无法正常响应,就会抛出异常,我们便可以进行异常处理
producer.send(record).get();

异步发送

producer.send(record, new Callback() 
      public void onCompletion(RecordMetadata metadata, Exception exception) 
      if (exception == null) 
         System.out.println(metadata.partition() + ":" + metadata.offset());
      
   
);

1.4 序列化器

消息要到网络上进行传输,必须进行序列化,而序列化器的作用就是如此。

Kafka 提供了默认的字符串序列化器(org.apache.kafka.common.serialization.StringSerializer),还有整型(IntegerSerializer)和字节数组(BytesSerializer)序列化器,这些序列化器都实现了接口(org.apache.kafka.common.serialization.Serializer)基本上能够满足大部分场景的需求。

1.5 自定义序列化器

见代码库:com.heima.kafka.chapter2.CompanySerializer

/** 
 * 自定义序列化器 
   @Override 
   public void configure(Map configs, boolean isKey)  
   
 
   @Override 
   public byte[] serialize(String topic, Company data)  
       if (data == null)  
           return null; 
    
   byte[] name, address; 
   try  
       if (data.getName() != null)  
           name = data.getName().getBytes("UTF-8"); 
    else  
      name = new byte[0]; 
    
   if (data.getAddress() != null)  
       address = data.getAddress().getBytes("UTF-8"); 
    else  
   address = new byte[0]; 
    
   ByteBuffer buffer = ByteBuffer. 
            allocate(4 + 4 + name.length + address.length); 
   buffer.putInt(name.length); 
   buffer.put(name); 
   buffer.putInt(address.length); 
   buffer.put(address); 
   return buffer.array(); 
 catch (UnsupportedEncodingException e)  
      e.printStackTrace(); 
    
   return new byte[0]; 
 

   @Override 
   public void close()  
    

使用自定义的序列化器

见代码库:com.heima.kafka.chapter2.ProducerDefifineSerializer

public class ProducerDefineSerializer 
    public static final String brokerList = "localhost:9092";
    public static final String topic = "heima";

    public static void main(String[] args)
            throws ExecutionException, InterruptedException 
        Properties properties = new Properties();
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                CompanySerializer.class.getName());
        // properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        KafkaProducer<String, Company> producer =
                new KafkaProducer<>(properties);
        Company company = Company.builder().name("kafka")
                .address("北京").build();
//                 Company company = Company.builder().name("hiddenkafka")
//                 .address("China").telphone("13000000000").build();
        ProducerRecord<String, Company> record =
                new ProducerRecord<>(topic, company);
        producer.send(record).get();
        

1.6 分区器

本身kafka有自己的分区策略的,如果未指定,就会使用默认的分区策略:

Kafka根据传递消息的key来进行分区的分配,即hash(key) % numPartitions。如果Key相同的话,那么就会分配到统一分区。

源代码org.apache.kafka.clients.producer.internals.DefaultPartitioner分析

自定义分区器见代码库 com.heima.kafka.chapter2.DefifinePartitioner

/**
  * 自定义分区器
  */
public class DefinePartitioner implements Partitioner 
    private final AtomicInteger counter = new AtomicInteger(0);

@Override
    public int partition(String topic, Object key, byte[] keyBytes,
    int numPartitions = partitions.size();
    if (null == keyBytes) 
        return counter.getAndIncrement() % numPartitions;
     else
       return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    
    @Override
    public void close() 
    
    @Override
    public void configure(Map<String, ?> configs) 
    

实现自定义分区器需要通过配置参数ProducerConfifig.PARTITIONER_CLASS_CONFIG来实现

// 自定义分区器的使用
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,DefinePartitioner.class.getNam
e());

1.7 拦截器

Producer拦截器(interceptor)是个相当新的功能,它和consumer端interceptor是在Kafka 0.10版本被引入的,主要用于实现clients端的定制化控制逻辑。

生产者拦截器可以用在消息发送前做一些准备工作。

使用场景

  • 1、按照某个规则过滤掉不符合要求的消息
  • 2、修改消息的内容
  • 3、统计类需求

见代码库:自定义拦截器com.heima.kafka.chapter2.ProducerInterceptorPrefifix

/**
  * 自定义拦截器
  */
 public class ProducerInterceptorPrefix implements
          ProducerInterceptor<String, String> 
      private volatile long sendSuccess = 0;
      private volatile long sendFailure = 0;

      @Override
      public ProducerRecord<String, String> onSend(
            ProducerRecord<String, String> record) 
         String modifiedValue = "prefix1-" + record.value();
         return new ProducerRecord<>(record.topic(),
            record.partition(), record.timestamp(),
            record.key(), modifiedValue, record.headers());
//             if (record.value().length() < 5) 
//             throw new RuntimeException();
   

      @Override
      public void onAcknowledgement(
            RecordMetadata recordMetadata,
            Exception e) 
      if (e == null) 
          sendSuccess++;
       else 
            sendFailure++;
      


      @Override
      public void close() 
            double successRatio = (double) sendSuccess / (sendFailure +sendSuccess);
            System.out.println("[INFO] 发送成功率="+ String.format("%f", successRatio * 100) + "%");
      

      @Override
      public void configure(Map<String, ?> map) 
      

实现自定义拦截器之后需要在配置参数中指定这个拦截器,此参数的默认值为空,如下:

// 自定义拦截器使用
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,ProducerDefineSerializer.cla
ss.getName());

功能演示:

发送端

接收端

二、发送原理剖析

消息发送的过程中,涉及到两个线程协同工作,主线程首先将业务数据封装成ProducerRecord对象,之后调用send()方法将消息放入RecordAccumulator(消息收集器,也可以理解为主线程与Sender线程直接的缓冲区)中暂存,Sender线程负责将消息信息构成请求,并最终执行网络I/O的线程,它从RecordAccumulator中取出消息并批量发送出去,需要注意的是,KafkaProducer是线程安全的,多个线程间可以共享使用同一个KafkaProducer对象。

三、其他生产者参数

之前提及的默认三个客户端参数,大部分参数都有合理的默认值,一般情况下不需要修改它们,

参考官网:http://kafka.apache.org/documentation/#producerconfifigs

3.1 acks

这个参数用来指定分区中必须有多少个副本收到这条消息,之后生产者才会认为这条消息时写入成功的。acks是生产者客户端中非常重要的一个参数,它涉及到消息的可靠性和吞吐量之间的权衡。

  • ack=0, 生产者在成功写入消息之前不会等待任何来自服务器的相应。如果出现问题生产者是感知不到的,消息就丢失了。不过因为生产者不需要等待服务器响应,所以它可以以网络能够支持的最大速度发送消息,从而达到很高的吞吐量。
  • ack=1,默认值为1,只要集群的首领节点收到消息,生产这就会收到一个来自服务器的成功响应。如果消息无法达到首领节点(比如首领节点崩溃,新的首领还没有被选举出来),生产者会收到一个错误响应,为了避免数据丢失,生产者会重发消息。但是,这样还有可能会导致数据丢失,如果收到写成功通知,此时首领节点还没来的及同步数据到follower节点,首领节点崩溃,就会导致数据丢失。
  • ack=-1,只有当所有参与复制的节点都收到消息时,生产这会收到一个来自服务器的成功响应,这种模式是最安全的,它可以保证不止一个服务器收到消息。

注意:acks参数配置的是一个字符串类型,而不是整数类型,如果配置为整数类型会抛出以下异常

3.2 retries

生产者从服务器收到的错误有可能是临时性的错误(比如分区找不到首领)。在这种情况下,如果达到了 retires 设置的次数,生产者会放弃重试并返回错误。默认情况下,生产者会在每次重试之间等待100ms,可以通过 retry.backoff.ms 参数来修改这个时间间隔。

3.3 batch.size

当有多个消息要被发送到同一个分区时,生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小,按照字节数计算,而不是消息个数。当批次被填满,批次里的所有消息会被发送出去。不过生产者并不一定都会等到批次被填满才发送,半满的批次,甚至只包含一个消息的批次也可能被发送。所以就算把 batch.size 设置的很大,也不会造成延迟,只会占用更多的内存而已,如果设置的太小,生产者会因为频繁发送消息而增加一些额外的开销。

3.4 max.request.size

该参数用于控制生产者发送的请求大小,它可以指定能发送的单个消息的最大值,也可以指单个请求里所有消息的总大小。 broker 对可接收的消息最大值也有自己的限制( message.max.size ),所以两边的配置最好匹配,避免生产者发送的消息被 broker 拒绝。

总结

本章主要讲了生产者客户端的用法以及整体流程架构,主要内容包括配置参数的详解、消息的发送方式、序列化器、分区器、拦截器等,在实际使用中,Kafka已经提供了良好的Java客户端支持,提高了开发效率。

以上是关于一文弄懂如何创建并使用Kafka生产者的主要内容,如果未能解决你的问题,请参考以下文章

kafka极简入门(三)--创建topic

一文精通kafka 消费者的三种语义

一文精通kafka 消费者的三种语义

一文让你彻底弄懂MySQL自增列

一文读懂kafka的幂等生产者

spring kafka 参数说明