从零搭建开发脚手架 Spring Boot集成Kafka实现生产者消费者的多种方式

雨点打透心脏的1/2处 2022-10-17 10:44 263阅读 0赞

### 文章目录 ###

*   *  Kafka简介和安装
     *  Spring Boot集成
     *   *  pom
         *  配置
     *  创建Kafka主题
     *  生产者
     *   *  简单发布消息
         *  带有回调发布消息
     *  消费者
     *   *  简单消费者
         *  消费特定分区的消息
         *  为监听器添加消息过滤器
     *  自定义消息转换器
     *   *  发送自定义消息
         *  消费自定义消息
     *  JavaConfig方式的生产者、消费者
     *   *  生产者
         *  消费者

## Kafka简介和安装 ##

kafka是一个分布式消息队列。在项目中应用十分广泛，具有高性能、持久化、多副本备份、横向扩展能力。

[官网][Link 1]戳这里。  
下载并解压（注意需要Kafka与Spring Boot版本对应，可以参考[这里][Link 2]）

tar -xvf kafka_2.12-2.5.0.tgz
    cd kafka_2.12-2.5.0

接着启动ZooKeeper与Kafka：

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
    bin/kafka-server-start.sh config/server.properties

Kafka需要用到ZooKeeper，需要在启动Kafka之前启动ZooKeeper（ZooKeeper是一个开源的分布式应用程序协调服务，是Hadoop的组件，主要用于解决分布式应用中的一些数据管理问题）。  
Kafka默认使用9092端口，部署在服务器上需要注意防火墙以及安全组的处理。

**相关术语**

*  Broker：Kafka集群包含一个或多个服务器，这种服务器称为Broker
 *  Topic：每条发布到Kafka的消息都有一个类别，这个类别称为Topic。物理上不同Topic的消息分开存储，逻辑上Topic的消息虽然保存在一个或多个Broker上，但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存放于何处
 *  Partition：每个Topic包含一个或多个Partition
 *  Producer：生产者，负责发布消息到Broker
 *  Consumer：消费者，向Broker读取消息的客户端
 *  Consumer Group：每个Consumer属于一个特定的Consumer Group，可以为每个Consumer指定Group Name，否则属于默认Group。

## Spring Boot集成 ##

### pom ###

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

### 配置 ###

spring:
      kafka:
        # 配置kafka broker地址 多个用 , 分割
        bootstrap-servers: 192.16.1.101:9092
        # 发出请求时传递给服务器的标识。用于服务器端日志记录。
        client-id: laker1
        # 生产者配置
        producer:
          # 发生错误后，消息重发的次数。
          retries: 2
          #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
          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
          # 生产者生成的所有数据的压缩类型。
          # compression-type:
        # 消费者配置 
        consumer:
          # 自动提交的时间间隔 在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
          # 如果没有足够的数据立即满足“fetch-min-size”给出的要求，服务器在响应获取请求之前阻塞的最长时间
          # fetch-max-wait:
          # 服务器应为获取请求返回的最小数据量。
          # fetch-min-size:
          # 单次调用 poll() 时返回的最大记录数。
          #max-poll-records:
          # 到消费者协调器的心跳之间的预期时间。
          #heartbeat-interval:
          # 读取以事务方式写入的消息的隔离级别。默认读未提交
          #isolation-level: read_committed
        listener:
          # 在侦听器容器中运行的线程数。
          concurrency: 5
          # listner负责ack，每调用一次，就立即commit AckMode枚举都在这里
          ack-mode: manual_immediate
          # 如果代理上不存在至少一个配置的主题，容器是否应该无法启动。默认false
          missing-topics-fatal: false
          # 当 ackMode 为“COUNT”或“COUNT_TIME”时，偏移提交之间的记录数。
          #ack-count:
          # 当 ackMode 为“TIME”或“COUNT_TIME”时，偏移提交之间的时间。
          #ack-time:
          # 发布空闲消费者事件之间的时间（未收到数据）。
          #idle-event-interval:
          #是否在初始化期间记录容器配置（INFO 级别）。
          #log-container-config:
          # 检查无响应消费者的时间间隔。 如果未指定持续时间后缀，则将使用秒。
          #monitor-interval:
          # 乘数应用于“pollTimeout”以确定消费者是否无响应。
          #no-poll-threshold:
          #轮询消费者时使用的超时。
          #poll-timeout:
          #侦听器类型。默认single，可选batch
          #type: single

## 创建Kafka主题 ##

**方式一（不推荐）**: 自动创建主题

在配置文件里指定好kafka的topic之后，调用send方法或者KafkaListener指定topic会自动帮我们创建好topic，只是创建的topic默认是1个副本和1个分区的，这一般不能满足我们的要求，所以我们还需要在kafka的server.properties里增加或修改以下参数：

auto.create.topics.enable=true
    num.partitions=3
    default.replication.factor=3

之后，kafka自动帮我们创建的主题都会包含3个副本和3个分区。

**方式二：可以提前**运行命令行工具在 Kafka 中创建主题：

$ bin/kafka-topics.sh --create \
      --zookeeper localhost:2181 \
      --replication-factor 1 --partitions 1 \
      --topic mytopic

**方式三**：随着Kafka中*AdminClient*的引入，我们现在可以以编程方式创建主题。

我们需要添加**KafkaAdmin Spring bean**，它将自动为**NewTopic**类型的所有 bean 添加主题：

@Configuration
    public class KafkaTopicConfig { 
        @Value(value = "${kafka.bootstrapAddress}")
        private String bootstrapAddress;
        @Bean// 使用 Spring Boot 时，KafkaAdmin会自动注册一个bean，因此您只需要NewTopic @Bean
        public KafkaAdmin kafkaAdmin() { 
            Map<String, Object> configs = new HashMap<>();
            configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
            return new KafkaAdmin(configs);
        }
        @Bean
        public NewTopic topic1() { 
             // 主题名称、分区数、副本数 或者使用TopicBuilder
             return new NewTopic("baeldung", 1, (short) 1);
        }
    }

## 生产者 ##

@Configuration
    public class KafkaProducerConfig { 
        @Bean
        public ProducerFactory<String, String> producerFactory() { 
            Map<String, Object> configProps = new HashMap<>();
            configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapAddress);
            configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
            configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
            return new DefaultKafkaProducerFactory<>(configProps);
        }
        @Bean
        public KafkaTemplate<String, String> kafkaTemplate() { 
            return new KafkaTemplate<>(producerFactory());
        }
    }

### 简单发布消息 ###

我们可以使用*KafkaTemplate*类发送消息：

@Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendMessage(String msg) { 
        kafkaTemplate.send(topicName, msg);
    }

### 带有回调发布消息 ###

在发送API返回的ListenableFuture对象可以阻塞发送线程并获取发送消息的结果，线程将等待结果，但它会减慢生产者的速度。

Kafka 是一个快速的流处理平台。因此，最好异步处理结果，以便后续消息不会等待上一条消息的结果。

我们可以通过回调来做到这一点：

public void sendMessage(String message) {        
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, message);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() { 
            // 发送成功的处理
            @Override
            public void onSuccess(SendResult<String, String> result) { 
                System.out.println("Sent message=[" + message + 
                  "] with offset=[" + result.getRecordMetadata().offset() + "]");
            }
            @Override
            public void onFailure(Throwable ex) { 
                // 发送失败的处理
                System.out.println("Unable to send message=[" 
                  + message + "] due to : " + ex.getMessage());
            }
        });
    }

## 消费者 ##

### 简单消费者 ###

@KafkaListener(topics = "topicName", groupId = "foo")
    public void listenGroupFoo(String message) { 
        System.out.println("Received Message in group foo: " + message);
    }

**我们可以为一个主题实现多个侦听器**，每个**侦听器**都有不同的组 ID。此外，一个消费者可以监听来自不同主题的消息：

@KafkaListener(topics = "topic1, topic2", groupId = "foo")

Spring 还支持在侦听器中使用[@Header][Header]注释检索一个或多个消息头：

@KafkaListener(topics = "topicName")
    public void listenWithHeaders(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) { 
          System.out.println("Received Message: " + message" + "from partition: " + partition);
    }

`@KafkaListener`可以接受的参数有：

*  data ： 对于data值的类型其实并没有限定，根据KafkaTemplate所定义的类型来决定。 data为List集合的则是用作批量消费。
 *  ConsumerRecord：具体消费数据类，包含Headers信息、分区信息、时间戳等
 *  Acknowledgment：用作Ack机制的接口
 *  Consumer：消费者类，使用该类我们可以手动提交偏移量、控制消费速率等功能

public void listen1(String data) 
    public void listen2(ConsumerRecord<K,V> data) 
    public void listen3(ConsumerRecord<K,V> data, Acknowledgment acknowledgment) 
    public void listen4(ConsumerRecord<K,V> data, Acknowledgment acknowledgment, Consumer<K,V> consumer) 
    public void listen5(List<String> data) 
    public void listen6(List<ConsumerRecord<K,V>> data) 
    public void listen7(List<ConsumerRecord<K,V>> data, Acknowledgment acknowledgment) 
    public void listen8(List<ConsumerRecord<K,V>> data, Acknowledgment acknowledgment, Consumer<K,V> consumer)

### 消费特定分区的消息 ###

对于具有多个分区的主题，`@KafkaListener`可以显式订阅具有初始偏移量的主题的特定分区：

@KafkaListener(
      topicPartitions = @TopicPartition(topic = "topicName",
      partitionOffsets = { 
        @PartitionOffset(partition = "0", initialOffset = "0"), 
        @PartitionOffset(partition = "3", initialOffset = "0")}),
      containerFactory = "partitionsKafkaListenerContainerFactory")
    public void listenToPartition(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) { 
          System.out.println("Received Message: " + message" + "from partition: " + partition);
    }

由于此侦听器中的`initialOffset`已设置为 0，因此每次初始化此侦听器时，将重新使用来自分区 0 和 3 的所有先前消耗的消息。

如果我们不需要设置偏移量，我们可以使用`@TopicPartition`注解的`partitions`属性，只设置没有偏移量的分区：

@KafkaListener(topicPartitions 
      = @TopicPartition(topic = "topicName", partitions = {  "0", "1" }))

### 为监听器添加消息过滤器 ###

我们可以通过添加自定义过滤器来配置侦听器以使用特定类型的消息。这可以通过将[RecordFilterStrategy][]设置到[KafkaListenerContainerFactory][RecordFilterStrategy]来完成：

@Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> filterKafkaListenerContainerFactory() { 
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setRecordFilterStrategy(record -> record.value().contains("World"));
        return factory;
    }

然后我们可以配置一个监听器来使用这个容器工厂：

@KafkaListener(topics = "topicName", containerFactory = "filterKafkaListenerContainerFactory")
    public void listenWithFilter(String message) { 
        System.out.println("Received Message in filtered listener: " + message);
    }

在这个监听器中，所有**匹配过滤器**的**消息都将被丢弃。**

## 自定义消息转换器 ##

前面消息发送和接收都是字符串。但是，我们也可以发送和接收自定义 Java 对象。这需要在`ProducerFactory`中配置适当的序列化器，在`ConsumerFactory`中配置反序列化器。

让我们看一个简单的 bean 类\*，\*我们将其作为消息发送：

public class Greeting { 
        private String msg;
        private String name;
    }

### 发送自定义消息 ###

我们使用[JsonSerializer][]。

`ProducerFactory`和`KafkaTemplate`的代码：

@Bean
    public ProducerFactory<String, Greeting> greetingProducerFactory() { 
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }
    @Bean
    public KafkaTemplate<String, Greeting> greetingKafkaTemplate() { 
        return new KafkaTemplate<>(greetingProducerFactory());
    }

我们可以使用这个新的`KafkaTemplate`来发送`Greeting`消息：

kafkaTemplate.send(topicName, new Greeting("Hello", "World"));

### 消费自定义消息 ###

同样，让我们修改`ConsumerFactory`和`KafkaListenerContainerFactory`以正确反序列化 Greeting 消息：

@Bean
    public ConsumerFactory<String, Greeting> greetingConsumerFactory() { 
        // ...
        return new DefaultKafkaConsumerFactory<>(props,new StringDeserializer(), new JsonDeserializer<>(Greeting.class));
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Greeting> greetingKafkaListenerContainerFactory() { 
        ConcurrentKafkaListenerContainerFactory<String, Greeting> factory =new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(greetingConsumerFactory());
        return factory;
    }

spring-kafka JSON 序列化器和反序列化器使用[Jackson][]库，它也是 spring-kafka 项目的可选 Maven 依赖项。

所以，让我们将它添加到我们的*pom.xml 中*：

<dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>

监听器代码：

@KafkaListener(
      topics = "topicName", 
      containerFactory = "greetingKafkaListenerContainerFactory")
    public void greetingListener(Greeting greeting) { 
        // process greeting message
    }

## JavaConfig方式的生产者、消费者 ##

@Configuration
    public class Kafka_Config { 
    
        @Value("${kafka.broker.list}")
        public String brokerList;
    
        public static final String topic = "TOPIC_LIN_LIANG";
    
        public final String groupId = "group.01";
    
        public Properties customerConfigs() { 
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);//自动位移提交
            props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);//自动位移提交间隔时间
            props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);//消费组失效超时时间
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");//位移丢失和位移越界后的恢复起始位置
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                    StringDeserializer.class.getName());
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                    StringDeserializer.class.getName());
    
            return props;
        }
    
        public Properties producerConfigs() { 
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
            props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 20000000);//20M 消息缓存
            //生产者空间不足时，send()被阻塞的时间，默认60s
            props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 6000);
            //生产者重试次数
            props.put(ProducerConfig.RETRIES_CONFIG, 0);
            //指定ProducerBatch（消息累加器中BufferPool中的）可复用大小
            props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
            //生产者会在ProducerBatch被填满或者等待超过LINGER_MS_CONFIG时发送
            props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer.client.id.demo");
            return props;
        }
    
    
        @Bean
        public Producer<Integer, Object> getKafkaProducer() { 
            //KafkaProducer是线程安全的，可以在多个线程中共享单个实例
            return new KafkaProducer<Integer, Object>(producerConfigs());
        }
    
    }

### 生产者 ###

@Component
    public class Kafka_Producer  { 
    
        public String topic = Kafka_Config.topic;
    
        @Autowired
        Producer producer;
    
        public void producer() throws Exception { 
    
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, "hello, Kafka!");
            try { 
                producer.send(record, new Callback() { 
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) { 
                        if (exception == null) { 
                            System.out.println(metadata.partition() + ":" + metadata.offset());
                        }
                    }
                });
            } catch (Exception e) { 
            }
        }
    }

### 消费者 ###

@Component
    public class Kafka_Consumer implements InitializingBean { 
        public String topic = Kafka_Config.topic;
        @Autowired
        Kafka_Config kafka_config;
    
        @Override
        public void afterPropertiesSet() throws Exception { 
            //每个线程一个KafkaConsumer实例，且线程数设置成分区数，最大化提高消费能力
            int consumerThreadNum = 2;//线程数设置成分区数，最大化提高消费能力
            for (int i = 0; i < consumerThreadNum; i++) { 
                new KafkaConsumerThread(kafka_config.customerConfigs(), topic).start();
            }
        }
    
        public class KafkaConsumerThread extends Thread { 
            private KafkaConsumer<String, String> kafkaConsumer;
    
            public KafkaConsumerThread(Properties props, String topic) { 
                this.kafkaConsumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(props);
                this.kafkaConsumer.subscribe(Arrays.asList(topic));
            }
    
            @Override
            public void run() { 
                try { 
                    while (true) { 
                        ConsumerRecords<String, String> records =
                                kafkaConsumer.poll(Duration.ofMillis(100));
                        for (ConsumerRecord<String, String> record : records) { 
                            System.out.println("message------------ "+record.value());
                        }
                    }
                } catch (Exception e) { 
                } finally { 
                    kafkaConsumer.close();
                }
            }
        }
    }

**参考：**

*  https://www.jianshu.com/p/6ce5d9a96113
 *  https://www.baeldung.com/

[Link 1]: https://kafka.apache.org/quickstart
[Link 2]: https://spring.io/projects/spring-kafka
[Header]: https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/messaging/handler/annotation/Header.html
[RecordFilterStrategy]: https://docs.spring.io/spring-kafka/api/org/springframework/kafka/listener/adapter/RecordFilterStrategy.html
[JsonSerializer]: https://docs.spring.io/spring-kafka/api/org/springframework/kafka/support/serializer/JsonSerializer.html
[Jackson]: https://www.baeldung.com/jackson