摘要

对于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依赖如下：

        org.springframework.bootspring-boot-starter-web2.7.9org.springframework.kafkaspring-kafka2.9.6

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

server:port: 8087
spring:mvc:pathmatch:matching-strategy: ant_path_matcherkafka:bootstrap-servers: 192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094consumer: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;/*** 
* 测试 前端控制器* 
* @author LongQi* @since 2021-06-23*/@Slf4j
@RestController
@RequestMapping("/test")
@Api(value = "TestController", tags = {"测试 API"})
public class TestController {@Autowiredprivate KafkaTemplate 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、添加依赖，具体如下：

        org.apache.kafkakafka-clients3.4.0

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() {@Overridepublic void run() {while (true){ConsumerRecords records = kafkaConsumer.poll(Duration.ofMillis(10000));for(ConsumerRecord 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()));}
}

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

成功接收到消息并打印