摘要

对于java的kafka集成，一般选用springboot集成kafka，但可能由于对接方kafka老旧、kafka不安全等问题导致kafak版本与spring版本不兼容，这个时候就得自己根据kafka客户端api集成了。

一、springboot集成kafka

具体官方文档地址：https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/

1、加入依赖，spring-boot-starter-web和spring-kafka 的版本号可以看它们依赖的spring版本是否一致，这里pom依赖如下：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.7.9</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.9.6</version>
        </dependency>

2、添加application.yml配置,具体如下：

server:
  port: 8087
spring:
  mvc:
    pathmatch:
      matching-strategy: ant_path_matcher
  kafka:
    bootstrap-servers: 192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094
    consumer:
      properties:
        group:
          id: boot-kafka

3、发送消息，由于KafkaTemplate是自动装配的，所以只要在spring的bean里注入KafkaTemplate发送消息即可，具体如下：

package com.longqi.bootkafka.controller;

import com.longqi.bootkafka.entity.MessageParam;
import com.longqi.bootkafka.entity.Wrapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.validation.Valid;

/**
 * <p>
 * 测试 前端控制器
 * </p>
 * @author LongQi
 * @since 2021-06-23
 */

@Slf4j
@RestController
@RequestMapping("/test")
@Api(value = "TestController", tags = {"测试 API"})
public class TestController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private Boolean isSend = true;

    @PostMapping("/kafka/sendMessage")
    @ApiOperation(httpMethod = "POST", value = "发送kafka告警消息", response = Wrapper.class)
    public Wrapper sendKafkaMessage(@Valid @ApiParam("参数") @RequestBody MessageParam param) {
        kafkaTemplate.send(param.getTopic(), param.getMessage());
        return Wrapper.ok(true);
    }

}

这里用参数{"message": "asd54a6d46a4ds","topic": "device-alarm-test"}进行测试，会报如下日志：

发现会报警告：[Producer clientId=producer-1] Error while fetching metadata with correlation id 34 : {device-alarm-test=LEADER_NOT_AVAILABLE}，获取主题元数据错误，这个可以忽略，查找元数据失败，kafka默认会自动创建主题的，后续再次发送消息，是不会报这个错误的。

查看可视化工具EFAK，发现主题device-alarm-test是自动创建成功，分区数是kafka的集群配置service.properties里配置的分区9，具体如下：

可以看到，其中一个分区保存了这个消息，logsize变成了1，说明这个消息是发送成功的。另外也可以看到主题的各分区主备消息所在的节点是不一样的。

4、接收消息，接收消息也很简单，只要在spring的bean里使用KafkaListener注解即可，具体如下：

可视化工具也能看到该主题该消费者9个分区的消费情况，具体如下：

logSize为存入分区parttion消息数量，Offset为消费的偏移量(已消费的数量)，Lag为未消费的数量(积压的数量)，Owner为消费者，目前可以看到消费者为同一个，即只有1个线程在消费这9个分区的消息。

二、客户端集成kafka

直接使用kafka客户端，建议使用最新版的客户端，毕竟没有其他框架版本限制，能用最新的就用最新的，毕竟新的一般性能强也修复了bug。好比23年2月份出现的kafka安全漏洞：远程代码执行漏洞CVE-2023-25194，对现在最新版3.4.0无效，对以前大部分版本就有效。

1、添加依赖，具体如下：

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

2、发送和消费消息，具体代码如下：

package com.longqi.bootkafka.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author LongQi
 * @projectName boot-integration
 * @description: kafka配置
 * @date 2023/3/13 14:42
 */

public class KafkaConfig {

    public static void main(String[] args) {
        // 声明主题
        String topic = "device-alarm-test";
        // 创建消费者
        Properties consumerConfig = new Properties();
        consumerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094");
        consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG,"boot-kafka");
        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer kafkaConsumer = new KafkaConsumer(consumerConfig);
        // 订阅主题并循环拉取消息
        kafkaConsumer.subscribe(Arrays.asList(topic));
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(10000));
                    for(ConsumerRecord<String, String> record:records){
                        System.out.println(record.value());
                    }
                }
            }
        }).start();
        // 创建生产者
        Properties producerConfig = new Properties();
        producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094");
        producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG,"boot-kafka-client");
        producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer producer = new KafkaProducer<>(producerConfig);
        // 给主题发送消息
        producer.send(new ProducerRecord<>(topic, "hello，"+System.currentTimeMillis()));
    }
}

最后可以看到打印消息如下：

成功接收到消息并打印