Kafka的JavaAPI

针对java程序员来说，必定是在代码中对kafka进行操作。

1.未整合版的Kafka的API

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

1.1 Producer 消息发送端代码

package cn.zl.springbootdemo.kafkaclient;

import cn.zl.springbootdemo.model.City;
import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

/**
 * @ClassName : MsgProducer
 * @Description :
 * @Date: 2021-07-21 17:54
 */
public class MsgProducer {
    // 生产者主题
    private final static String TOPIC_NAME = "my-replicated-topic";
    // 生产者的配置
    private final static Properties props = new Properties();
    // 生产者的对象。
    private static KafkaProducer<String, String> producer = null;

    /**
     * 基础配置,满足这几个配置我们的producer就可以生产数据了。
     */
    public static void baseConfiguration() {
        // 配置kafka的IP，如果是集群需要配置多个IP
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        // 配置key的序列化，key的作用用来计算分区
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 配置value的序列化，value就是真正要传输的数据
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    }

    /**
     * 进阶配置
     */
    public static void advancedConfiguration() {
        /** acks消息确认机制，即消息是否发到了副本上，或者已经同步到几个副本上。
         * 1.acks=0 ：表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送一条消息。性能最高，但是最容易丢失消息。
         * 2.acks=1 ：至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成写入。就可以继续发送下一条消息。
         *          这种情况下，如果follower没有成功备份数据，而此时leader又挂掉了，则消息会丢失。
         * 3.acks=-1或all ：需要等待 min.insync.replicas(默认为1，推荐配置大于等于2)这个参数配置的副本个数都成功写入日志，这种策略
         *          或保证只要有一个备份就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或者跟钱打交道的场景才会使用这种配置。
         */
        props.put(ProducerConfig.ACKS_CONFIG, "1");

        /**
         * 发送失败会重试，默认重试间隔100ms，重试能够保证消息发送的可靠性，但是也可能造成消息重复发送，比如网络抖动，所以需要在接受者那边做好
         * 消息接收的幂等性处理
         */
        props.put(ProducerConfig.RETRIES_CONFIG, 3);

        /**
         * 重试间隔设置,比如说此处300，毫秒级别。
         */
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,300);

        /**
         * 设置发送消息的本地缓存区，如果设置该缓存区，消息会先发送到本地缓存区，可以提高消息发送性能，默认值是33554432，即32MB
         */
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);

        /**
         * kafak本地线程会从缓冲区，批量发送到broker，设置批量发送的消息的大小，默认值为16384，也就是说一个batch满了16KB就发送出去。
         */
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);

        /**
         * 消息发送的时间差，默认值为0，意思就是消息必须立即被发送，但是这样会影响性能
         * 一般设置为10毫秒左右，就是说这个消息发送完后会进入本地的一本batch，如果10毫秒内，这个batch满了16kb就会随batch一起被发送出去
         * 如果10毫秒内，batch没满，那么久必须把消息发送出去，不能让消息的发送延迟时间太长
         */
        props.put(ProducerConfig.LINGER_MS_CONFIG,10);
    }

    /**
     * 同步发送版本,此处以循环发送5条数据为例
     */
    private static void syncSend() throws ExecutionException, InterruptedException {
        // 创建消费者
        producer = new KafkaProducer<>(props);
        City citySync = new City();
        for (int i = 0; i < 5; i++) {
            citySync.setId(i);
            citySync.setName("test同步" + i);
            citySync.setCity_code(11111 + i);
            citySync.setProvince_code(22222 + i);
            // 封装成一条记录
            ProducerRecord<String, String> producerRecord = new ProducerRecord(TOPIC_NAME, String.valueOf(citySync.getId()), JSON.toJSONString(citySync));
            // 发送，阻塞原因是Future 的get。
            RecordMetadata recordMetadata = (RecordMetadata) producer.send(producerRecord).get();
            System.out.println("同步方式发送消息结果：" + "  topic:" + recordMetadata.topic() + "   partition" + recordMetadata.partition() + "  offset" + recordMetadata.offset());
        }
        producer.close();
    }

    /**
     * 异步发送版本，此处以循环发送5条数据为例
     */
    private static void asyncSend() throws InterruptedException {
        // 创建消费者
        producer = new KafkaProducer<>(props);
        City cityAsync = new City();
        int msgNum = 5;
        final CountDownLatch countDownLatch = new CountDownLatch(msgNum);
        for (int i = 0; i < 5; i++) {
            cityAsync.setId(i);
            cityAsync.setName("test异步" + i);
            cityAsync.setCity_code(11111 + i);
            cityAsync.setProvince_code(22222 + i);
            // 封装成一条记录
            ProducerRecord<String, String> producerRecord = new ProducerRecord(TOPIC_NAME, String.valueOf(cityAsync.getId()), JSON.toJSONString(cityAsync));
            producer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    // 此处此完成发送回调，可能存在发送成功和不成功。根据异常来判断。
                    if (exception == null && metadata != null) {
                        System.out.println("同步方式发送消息结果：" + "  topic:" + metadata.topic() + "   partition" + metadata.partition() + "  offset" + metadata.offset());
                    } else {
                        System.err.println("消息发送失败" + exception.getStackTrace());
                    }
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        producer.close();
    }

    //测试一把
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 基础配置
        baseConfiguration();
        // 进阶配置
        advancedConfiguration();
        // 同步发送
//        syncSend();
        // 异步发送
        asyncSend();
    }
}

1.2 Consumer 消息消费端代码

package cn.zl.springbootdemo.kafkaclient;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**
 * @ClassName : MsgConsumer
 * @Description :
 * @Date: 2021-07-21 22:14
 */
public class MsgConsumer {

    // 主题
    private final static String TOPIC_NAME = "my-replicated-topic";
    // 消费者组
    private final static String CONSUMER_GROUP_NAME = "testGroup";
    // 消费者的配置
    private final static Properties props = new Properties();
    // 消费者
    private static KafkaConsumer<String, String> consumer;

    /**
     * 基础配置，消费者的基础配置
     */
    private static void basicConfiguration() {
        /**
         * 连接kafkaServer，如果是集群的情况下，最好设置多个IP。
         */
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        /**
         * 设置消费者组，消费者组的概念很重要，一个partition 只能被某个消费者组内的某个消费者消费。
         */
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        /**
         * key的序列化，具体工作就是将字符串转化为字节数组
         */
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        /**
         * value的序列化，具体工作就是将字符串转化为字节数组
         */
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    }

    /**
     * 进阶配置
     */
    public void advancedConfiguration() {
        /**
         * 是否自动提交offset，默认就是true,offset表示我们消费到哪里了。
         */
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        /**
         * 自动提交offset的间隔时间
         */
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        /**
         * 当消费主题是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费？
         * latest(默认)：只消费自己启动之后发送到主题的消息。
         * earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
         */
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        /**
         * 设置心跳检测时间，broker接受到心跳,如果此时有rebalance发生会通过心跳响应将rebalance方案下发consumer，这个时间可以稍微短一点。
         */
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        /**
         * 服务端broker 多久感知不到一个consumer心跳就认为他故障了，会将其提出消费组，对应的partition也会被重新分配给其他consumer，默认是10秒
         */
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
        /**
         * 一次poll最大拉取消息的条数，如果消费组处理速度很快，可以设置大点，如果处理速度一般，可以设置小点。
         */
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        /**
         * 如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太弱，会将其提出消费组，将分区分配给别的consumer消费
         */
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
    }

    /**
     * 此处只是将consumer的具体的消费代码提取出去，减少冗余代码，没其他深意。
     */
    private static void commonConsumer(){
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
            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());
            }
            // 手动同步提交offset，当前线程会阻塞直到offset提交成功
            // 一般使用同步提交，因为提交之后一般没有什么逻辑代码了
            consumer.commitSync();

//            // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面程序的逻辑
//            consumer.commitAsync(new OffsetCommitCallback() {
//                @Override
//                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
//                    // 一次异步提交。当exception没有异常的时候，就说明提交成功了。
//                    if (exception != null) {
//                        System.err.println("Commit failed for " + offsets);
//                        System.err.println("Commit failed exception: " + exception.getStackTrace());
//                    }
//                }
//            });
        }
    }

    /**
     * 普通的消费模式，就是启动一个消费者，生产消费者然后接受信息，支持订阅多个topic。
     * 此处没有指定partition，会去找这个主题下所有的partition，然后进行消费。支持顺序。
     */
    private static void generalConsumer() {
        // 创建kafka消费端
        consumer = new KafkaConsumer<String, String>(props);
        // 订阅主题,通过此处可以看出，传入的是个参数，所以可以传入多个topic。
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        // 调用消费方法。
        commonConsumer();
    }

    /**
     * 指定分区消费消息，无法保证有序性，除非只有一个分区。因为多个分区，写入的时候都不是有序的。
     */
    private static void specifiedPartitionConsumer() {
        // 创建kafka消费端
        consumer = new KafkaConsumer(props);
        // 指定partition 0
        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        // 调用消费方法。
        commonConsumer();
    }

    /**
     * 指定offset消费。
     */
    private static void specifiedOffsetConsumer(){
        // 创建kafka消费端
        consumer =  new KafkaConsumer(props);
        // 首先指定partition分区
        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
        // 将offset设置成5，表示从5开始消费。
        consumer.seek(new TopicPartition(TOPIC_NAME,0),5);
        commonConsumer();
    }


    /**
     * 回溯消费，假设我们服务宕机了，日志也有可能被清除了，我们并不知道当前消费到哪里了，然是我们依然想从头开始消费。
     * 一共有两种方式：
     *             1.获取当前分区，头的offset，然后通过seek方法来进行消费。
     *             2.直接调用seekToBeginning方法，自动识别头部，然后开始消费。
     */
    private static void backPartitionConsumer(){
        // 创建kafka消费端
        consumer =  new KafkaConsumer(props);
        // 首先指定partition分区
        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
        /**
         * 第一种：获取当前分区，头的offset，然后通过seek方法来进行消费。
         */
//        Map<TopicPartition, Long> topicPartitionLongMap = consumer.beginningOffsets(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
//        consumer.seek(new TopicPartition(TOPIC_NAME, 0), topicPartitionLongMap.get(new TopicPartition(TOPIC_NAME,0)));
        /**
         * 第二种：直接调用seekToBeginning方法，自动识别头部，然后开始消费。
         */
        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
        commonConsumer();
    }

    /**
     * 继续消费，假设我们服务宕机了，我们并不知道当前消费到哪里了，然是我们依然想从尾部继续消费。
     * 一共有两种方式：
     *              1. 获取当前分区，尾部的offset，通过seek方式来进行消费
     *              2. 直接调用seekToEnd方法，自动识别尾部，然后开始消费。
     */
    private static void continueConsume(){
        // 创建kafka消费端
        consumer =  new KafkaConsumer(props);
        // 首先指定partition分区
        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,0)));
        /**
         * 获取当前分区，尾部的offset，通过seek方式来进行消费
         */
//        Map<TopicPartition, Long> topicPartitionLongMap = consumer.endOffsets(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
//        consumer.seek(new TopicPartition(TOPIC_NAME, 0),topicPartitionLongMap.get(new TopicPartition(TOPIC_NAME, 0)));
        /**
         * 直接调用seekToEnd方法，自动识别尾部，然后开始消费。
         */
        consumer.seekToEnd(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        commonConsumer();
    }

    public static void main(String[] args) {
        // 基础配置
        basicConfiguration();
//        普通消费
//        generalConsumer();
//        指定分区消费
//        specifiedPartitionConsumer();
//        回溯消费
//        backPartitionConsumer();
//        指定offset消费
//        specifiedOffsetConsumer();
        continueConsume();
    }
}

2.Spring Boot整合Kafka

引入spring boot kafka依赖

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

application.yml配置如下：

server:
  port: 8080
spring:
### kafka配置
  kafka:
    bootstrap-servers: localhost:9092,localhost:9093,localhost:9093
    producer: #生产者
      # 发生错误后，消息重发的次数
      retries: 3
      # 当有多个小时需要被发送到统一分区时，生产者会把他们放在同一批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算
      batch-size: 16384
      # 设置生产者内存缓存区的大小
      buffer-memory: 33554432
      # 键的序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 值的序列化方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
      # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
      # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
      acks: 1
    consumer:
      group-id: default-group
      # 自动提交的时间间隔 在Spring Boot 2.x 版本中这里采用的值的类型Duration 需要符合特定的格式，如1S，1M，2H，5D
      auto-commit-interval: 1s
      # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
      # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
      # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
      auto-offset-reset: earliest
      # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
      enable-auto-commit: false
      # 键的反序列化方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 值的反序列化方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 当每一条记录被消费者监听器（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_IMMEDIATE
      # 在监听器容器中运行的线程数
      concurrency: 5
      #listner负责ack，每调用一次，就立即commit
      ack-mode: manual_immediate
      missing-topics-fatal: false

2.1 发送者代码(Producer)

package cn.zl.springbootdemo.controller.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @ClassName : KafkaController
 * @Description :
 * @Date: 2021-07-22 03:14
 */
@RestController("kafkaProducer")
public class KafkaController {

    private final static String TOPIC = "my-replicated-topic";

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @RequestMapping("/sendMsg")
    public void send(){
        kafkaTemplate.send(TOPIC,0,"key","this is a msg");
    }
}

2.2 发送者代码(Consumer)

package cn.zl.springbootdemo.controller.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @ClassName : MyConsumer
 * @Description :
 * @Date: 2021-07-22 03:17
 */
@Component
public class MyConsumer {
    /**
     * @KafkaListener(groupId = "testGroup", topicPartitions = {
     *             @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
     *             @TopicPartition(topic = "topic2", partitions = "0",
     *                     partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
     *     },concurrency = "6")
     *  //concurrency就是同组下的消费者个数，就是并发消费数，必须小于等于分区总数
     * @param record
     */
    @KafkaListener(topics = "my-replicated-topic",groupId = "listenGroupOne",topicPattern = "1")
    public void listenGroupOne(ConsumerRecord<String,String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        // 手动提交offset
        ack.acknowledge();
    }
    @KafkaListener(topics = "my-replicated-topic",groupId = "listenGroupTwo",topicPattern = "0")
    public void listenGroupTwo(ConsumerRecord<String,String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        // 手动提交offset
        ack.acknowledge();
    }
}