Kafka发送消息和读取消息

妖狐艹你老母 2023-06-23 13:21 1阅读 0赞

### 生产者发送消息到Kafka ###

我们先来看一个生产者-消费者最简单的实例.

public class DemoProducer {
    	public static void main(String[] args) {
    	    Properties props = new Properties();
    	    props.put("bootstrap.servers", "192.168.124.20:9092");
    	    props.put("acks", "all");
    	    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    	    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    	    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    	
    	    ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "greeting", "hello world! 520");
    	    //只发不管结果
    	    producer.send(record);
    	    producer.close();
    }
    
    public class DemoConsumer {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", "192.168.124.20:9092");
            props.setProperty("group.id", "test1");
            props.setProperty("enable.auto.commit", "false");
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Collections.singletonList("my-topic"));
    
            try {
                while (true) {
                    //poll参数值, 指明线程如果没有数据时等待多长时间，0表示不等待立即返回
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("offset = %d, key = %s, value = %s%n",record.offset(), record.key(), record.value());
                    }
                    consumer.commitSync();
                }
            }finally {
                consumer.close();
            }
        }
    }
    
    结果:
    offset = 0, key = greeting, value = hello world! 520

下面我们来分析下生产者发送消息过程.

**生产者的配置**

生产者有很多配置, 这里只介绍几个最简单的配置.

(1) bootstrap.servers

kafka服务地址, 只需要配置集群中的一个broker就行, 但保险起来一般配置多个, 以防某个broker宕机.

(2) acks

acks控制多少个分区副本必须写入消息后生产者才能认为写入成功，这个参数对消息丢失可能性有很大影响

*  acks=0：生产者把消息发送到broker即认为成功，不等待broker的处理结果. 这种方式的吞吐最高，但也是最容易丢失消息的.
 *  acks=1：生产者在该分区的主副本（leader）写入消息并返回成功后，认为消息发送成功. 这种方式能够一定程度避免消息丢失, 但如果在写入主副本后, 主副本还没有把消息同步到其他副本, 这时主副本所在broker宕机了, 那么该消息还是会丢失.
 *  acks=all：生产者会等待所有副本成功写入该消息, 这种方式是最安全的, 能够保证消息不丢失, 但是延迟也是最大的.

(3) key.serializer和value.serializer

配置key和value的序列化类, Kafka自带了两种序列化类, 一种是字节序列化类, 一种是字符串序列化类.

org.apache.kafka.common.serialization.BytesSerializer
    org.apache.kafka.common.serialization.StringSerializer

**创建消息**

上面的实例中, 我们创建了一个简单的String消息, 其中"my-topic"是主题名, "greeting"是key, "hello world! 520"是value.

ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "greeting", "hello world! 520");

在生产环境中, 我们一般会创建一个JSON消息, 然后以字节或字符串的形式发送到Kafka, 比如:

User user = new User();
    user.setId(1);
    user.setName("tyshawn");
    user.setAge(18);
    String msg = JSON.toJSONString(user);
    //可以不设置key
    ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", msg);
    producer.send(record);

**发送消息**

(1) 同步发送

在上面的实例中, 我们使用send()方法发送消息, 其实发送消息后会有一个返回值, send()方法的源码如下:

/**
     * Asynchronously send a record to a topic. Equivalent to <code>send(record, null)</code>.
     */
    @Override
    public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
        return send(record, null);
    }

我们可以把发送消息的代码块修改如下:

try {
        RecordMetadata recordMetadata = producer.send(record).get();
    } catch (Exception e) {
        //记录日志
        e.printStackTrace();
    }

注意, 这里使用了Future.get()来获取发送结果, 如果发送消息失败则会抛出异常，否则返回一个RecordMetadata对象. 在抛异常前, 生产者会进行重试, 如果重试超过一定次数仍不成功则抛出异常.

(2) 异步发送

当生产者与broker之间的网络延时比较大, 如果采用同步发送, 每条消息都需要等待比较长的时间, 这时我们可以采用异步发送的方式, 几乎没有任何耗时, 而且我们还可以通过回调知道消息的发送结果. 异步发送源码如下:

/**
     * Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.
     */
    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        // intercept the record, which can be potentially modified; this method does not throw exceptions
        ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
        return doSend(interceptedRecord, callback);
    }

我们可以把发送消息的代码块修改如下:

producer.send(record, new ProducerCallBack());
    
    public class ProducerCallBack implements Callback {
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e != null) {
                //记录日志
                e.printStackTrace();
            }
        }
    }

**自定义分配器**

默认情况下, 如果我们定义消息时没有设置key, Kafka会将消息均衡分配到所有的分区(轮询的方式), 如果设置了key, Kafka会对key进行哈希, 把得到的哈希值对分区数量进行取模, 最后得到的取模值决定了消息分派到哪个分区. Kafka除了默认分配器外, 我们用户也可以按照各自需求来自定义分配器.

自定义分配器需要实现org.apache.kafka.clients.producer.Partitioner接口, 源码如下:

public interface Partitioner extends Configurable, Closeable {
        /**
         * Compute the partition for the given record.
         *
         * @param topic The topic name
         * @param key The key to partition on (or null if no key)
         * @param keyBytes The serialized key to partition on( or null if no key)
         * @param value The value to partition on or null
         * @param valueBytes The serialized value to partition on or null
         * @param cluster The current cluster metadata
         */
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
    
        /**
         * This is called when partitioner is closed.
         */
        public void close();
    }
    
    public interface Configurable {
        /**
         * Configure this class with the given key-value pairs
         */
        void configure(Map<String, ?> configs);
    }

我们需要重写partition方法. 下面我们来自定义一个分配器, 所有key=order的记录存储在同一个分区, 其他记录采用key的哈希值取模的默认方式分区.

(1) 修改Kafka分区配置

# The default number of log partitions per topic. More partitions allow greater
    # parallelism for consumption, but this will also result in more files across
    # the brokers.
    num.partitions=3

(2) 自定义分区类

public class OrderPartitioner implements Partitioner {
        @Override
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
            List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
            int numPartitions = partitionInfos.size();
            if (key == null || !(key instanceof String)) {
                throw new InvalidRecordException("message key is not exist.");
            }
            if (key.equals("order")) {
                return numPartitions - 1;
            } else {
                return key.hashCode() % (numPartitions - 1);
            }
        }
    
        @Override
        public void close() {
        }
    
        @Override
        public void configure(Map<String, ?> configs) {
        }
    }

(3) 生产者

在属性配置中配置自定义分区.

(4) 消费者

(5) 结果

patitioner = 1, offset = 14, key = greeting, value = hello world! 520
    patitioner = 2, offset = 2, key = order, value = member order
    patitioner = 0, offset = 0, key = nickname, value = tyshawn

### 从Kafka中读取消息 ###

我们再来看下上面的消费者实例.

在属性配置中, bootstrap.servers与生产者配置一样, key.deserializer, value.deserializer是反序列化配置, group.id 用来指定消费者组, enable.auto.commit 用于指定消费者是否自动提交消费偏移量(offset), 默认为true.

上面的消费者实例中最核心的一行代码是consumer.poll(), 我们不断的调用poll()拉取数据, 如果停止拉取, 那么Kafka会认为此消费者已经死亡并进行重平衡. poll()方法其内部实现了协作、分区重平衡、心跳、数据拉取等功能, 但使用时这些细节都被隐藏了, 我们不需要关注这些.

需要注意一点, 消费者对象不是线程安全的, 也就是不能够多个线程同时使用一个消费者对象, 也不能够一个线程有多个消费者对象, 简而言之, 一个线程一个消费者, 如果需要多个消费者那么使用多线程来进行一一对应.

**提交（commit）与偏移量（offset）**

当我们调用poll()时, 该方法会返回我们订阅Topic中没有消费的消息. 当消费者消费完消息后, 需要调用commit()方法向Kafka发送分区中已经消费的消息偏移量, kafka进行记录. 当消费者宕机或新消费者加入时, Kafka会进行重平衡, 消费者会负责之前并不属于它的分区, 这个过程可能会导致消息丢失或消息重复消费.

> 消息丢失的情况: 在消息处理前先提交了偏移量, 如果这个时候进行了重平衡, 就会导致消息丢失.  
> 消息重复消费的情况: 在消息处理后再提交偏移量, 如果在消息处理过程中进行了重平衡, 就会导致重复消费.

在消息处理前后提交偏移量会导致不同的问题, 但我们一般都是在消息处理后再提交偏移量, 因为我们可以通过一系列方法尽量避免重复消费的情况, 下面就讲解这些方法.

(1) 自动提交偏移量

消费者会在poll()方法调用后每隔5秒（由auto.commit.interval.ms指定）提交一次偏移量, Kafka默认就是自动提交. 但这种方式可能会导致消息重复消费, 比如某个消费者poll消息后, 应用正在处理消息, 在3秒后Kafka进行了重平衡, 那么由于没有更新偏移量导致重平衡后这部分消息重复消费.

(2) 主动提交偏移量

为了减少消息重复消费, 我们会选择自己主动提交偏移量. 设置auto.commit.offset为false, 消息处理完后调用commitSync()来主动提交, 该方法会提交poll()返回的最后一个消息的偏移量. 我们的消费者实例就是使用的这种方式, 这种方式如果在处理消息时进行了重平衡, 会导致当前poll()消息重复消费. (比自动提交偏移量好一点)

(3) 异步提交偏移量

上面提到的commitSync()是同步提交, 同步提交有一个缺点, 那就是发起提交时应用会阻塞, 如果我们降低提交的频率, 当发生重平衡时, 重复消费的消息量又会增大, 所以我们一般使用异步提交的方式. 异步提交的缺点是, 如果Kafka返回提交失败, 同步提交会进行重试直到成功或者最后抛出异常给应用, 而异步提交不会进行重试. 所以使用异步提交时, 我们需要通过回调的方式记录提交结果.

consumer.commitAsync(new OffsetCommitCallback() {
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
            if (e != null) {
                //记录日志
            }
        }
    });

(4) 混合同步提交与异步提交

正常情况下, 偶然的提交失败并不是什么大问题, 只要后续的提交成功就可以了. 但是在某些情况下（例如程序退出、重平衡）, 我们希望最后的提交成功, 因此一种非常普遍的方式是混合异步提交和同步提交.

try {
        while (true) {
            //poll参数值, 指明线程如果没有数据时等待多长时间，0表示不等待立即返回
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("patitioner = %d, offset = %d, key = %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
            }
            //异步提交
            consumer.commitAsync();
        }
    } catch (Exception e) {
        //记录日志
    } finally {
        try {
            //同步提交
            consumer.commitSync();
        }finally {
            consumer.close();
        }
    }

(5) 提交特定的偏移量

commitSync()和commitAsync()会提交上一次poll()的最大偏移量, 但如果poll()返回消息数量非常多, 我们可能会希望在处理这些批量消息过程中提交位移, 以免发生重平衡时导致重复消费的消息过多. Kafka提交特定的偏移量源码如下:

/**
     * The committed offset should be the next message your application will consume,
     * i.e. lastProcessedMessageOffset + 1.
     */
    @Override
    public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback) {
        acquireAndEnsureOpen();
        try {
            maybeThrowInvalidGroupIdException();
            log.debug("Committing offsets: {}", offsets);
            offsets.forEach(this::updateLastSeenEpochIfNewer);
            coordinator.commitOffsetsAsync(new HashMap<>(offsets), callback);
        } finally {
            release();
        }
    }

注意一点, 提交特定的偏移量中的这个偏移量是上一次poll()的最大偏移量 + 1.

Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
    int count = 0;
    try {
        while (true) {
            //poll参数值, 指明线程如果没有数据时等待多长时间，0表示不等待立即返回
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("patitioner = %d, offset = %d, key = %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
    
                currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, ""));
                if (count % 1024 == 0) {
                    consumer.commitAsync(currentOffsets, null);
                }
                count++;
            }
        }
    } catch (Exception e) {
        //记录日志
    } finally {
        try {
            //同步提交
            consumer.commitSync(currentOffsets);
        }finally {
            consumer.close();
        }
    }

(6) 重平衡监听器（Rebalance Listener）

我们可以使用重平衡监听器, 在Kafka即将进行重平衡前将已经处理过的消息偏移量进行提交. Kafka提供了一个这样的接口ConsumerRebalanceListener, 我们只需要在调用subscribe()方法时传入实现了ConsumerRebalanceListener接口的对象, 其方法如下:

*  public void onPartitionRevoked(Collection partitions)：此方法会在消费者停止消费消费后，在重平衡开始前调用。
 *  public void onPartitionAssigned(Collection partitions)：此方法在分区分配给消费者后，在消费者开始读取消息前调用。
    
        public class DemoConsumer {
              private static Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
          
              public static void main(String[] args) {
                  Properties props = new Properties();
                  props.setProperty("bootstrap.servers", "192.168.124.20:9092");
                  props.setProperty("group.id", "test1");
                  props.setProperty("enable.auto.commit", "false");
                  props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
                  props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
                  KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
          
                  consumer.subscribe(Collections.singletonList("default-topic"), new RebalanceHandler(consumer));
                  try {
                      while (true) {
                          //poll参数值, 指明线程如果没有数据时等待多长时间，0表示不等待立即返回
                          ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                          for (ConsumerRecord<String, String> record : records) {
                              System.out.printf("patitioner = %d, offset = %d, key = %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
          
                              currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, ""));
                          }
                          consumer.commitAsync(currentOffsets, null);
                      }
                  } catch (Exception e) {
                      //记录日志
                  } finally {
                      try {
                          //同步提交
                          consumer.commitSync(currentOffsets);
                      }finally {
                          consumer.close();
                      }
                  }
              }
          
              static class RebalanceHandler implements ConsumerRebalanceListener {
                  private Consumer consumer;
          
                  public RebalanceHandler(Consumer consumer) {
                      this.consumer = consumer;
                  }
          
                  @Override
                  public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                      consumer.commitSync(currentOffsets);
                  }
          
                  @Override
                  public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                  }
              }
          }

**优雅退出**

在一般情况下, 我们会在一个主线程中循环poll消息并进行处理. 当需要退出poll循环时, 我们可以使用另一个线程调用consumer.wakeup(), 调用此方法会使得poll()抛出WakeupException. 如果调用wakup时, 主线程正在处理消息, 那么在下一次主线程调用poll时会抛出异常. 主线程在抛出WakeUpException后, 需要调用consumer.close(), 此方法会提交位移, 同时发送一个退出消费组的消息到Kafka. Kafka收到消息后会立即进行重平衡（无需等待此消费者会话过期）.

public class DemoConsumer {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", "192.168.124.20:9092");
            props.setProperty("group.id", "test1");
            props.setProperty("enable.auto.commit", "false");
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    
            consumer.subscribe(Collections.singletonList("default-topic"));
            try {
                while (true) {
                    //poll参数值, 指明线程如果没有数据时等待多长时间，0表示不等待立即返回
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("patitioner = %d, offset = %d, key = %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
                    }
                    consumer.commitAsync();
                }
            } catch (WakeupException e) {
                //记录日志
            } finally {
                try {
                    //同步提交
                    consumer.commitSync();
                }finally {
                    consumer.close();
                }
            }
    
            //注册JVM关闭时的回调钩子，当JVM关闭时调用此钩子。
            Runtime.getRuntime().addShutdownHook(new Thread(){
                public void run(){
                    consumer.wakeup();
                    try {
                        this.join();
                    } catch (InterruptedException e) {
                        //记录日志
                        e.printStackTrace();
                    }
                }
            });
        }
    }

看到这里我们应该知道, 虽然我们可以采用各种办法来尽量避免重复消费, 但重复消费是无法完全避免的, 这是消息队列的一个普遍问题. 我们只能在消费消息前, 先从数据库中判断该记录是否已被处理, 如果未被处理, 才继续消费.