摘要

對(duì)于java的kafka集成，一般選用springboot集成kafka，但可能由于對(duì)接方kafka老舊、kafka不安全等問(wèn)題導(dǎo)致kafak版本與spring版本不兼容，這個(gè)時(shí)候就得自己根據(jù)kafka客戶(hù)端api集成了。

一、springboot集成kafka

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

1、加入依賴(lài)，spring-boot-starter-web和spring-kafka 的版本號(hào)可以看它們依賴(lài)的spring版本是否一致，這里pom依賴(lài)如下：

        <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、發(fā)送消息，由于KafkaTemplate是自動(dòng)裝配的，所以只要在spring的bean里注入KafkaTemplate發(fā)送消息即可，具體如下：

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>
 * 測(cè)試 前端控制器
 * </p>
 * @author LongQi
 * @since 2021-06-23
 */

@Slf4j
@RestController
@RequestMapping("/test")
@Api(value = "TestController", tags = {"測(cè)試 API"})
public class TestController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private Boolean isSend = true;

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

}

這里用參數(shù){"message": "asd54a6d46a4ds","topic": "device-alarm-test"}進(jìn)行測(cè)試，會(huì)報(bào)如下日志：

發(fā)現(xiàn)會(huì)報(bào)警告：[Producer clientId=producer-1] Error while fetching metadata with correlation id 34 : {device-alarm-test=LEADER_NOT_AVAILABLE}，獲取主題元數(shù)據(jù)錯(cuò)誤，這個(gè)可以忽略，查找元數(shù)據(jù)失敗，kafka默認(rèn)會(huì)自動(dòng)創(chuàng)建主題的，后續(xù)再次發(fā)送消息，是不會(huì)報(bào)這個(gè)錯(cuò)誤的。

查看可視化工具EFAK，發(fā)現(xiàn)主題device-alarm-test是自動(dòng)創(chuàng)建成功，分區(qū)數(shù)是kafka的集群配置service.properties里配置的分區(qū)9，具體如下：

可以看到，其中一個(gè)分區(qū)保存了這個(gè)消息，logsize變成了1，說(shuō)明這個(gè)消息是發(fā)送成功的。另外也可以看到主題的各分區(qū)主備消息所在的節(jié)點(diǎn)是不一樣的。

4、接收消息，接收消息也很簡(jiǎn)單，只要在spring的bean里使用KafkaListener注解即可，具體如下：

可視化工具也能看到該主題該消費(fèi)者9個(gè)分區(qū)的消費(fèi)情況，具體如下：

logSize為存入分區(qū)parttion消息數(shù)量，Offset為消費(fèi)的偏移量(已消費(fèi)的數(shù)量)，Lag為未消費(fèi)的數(shù)量(積壓的數(shù)量)，Owner為消費(fèi)者，目前可以看到消費(fèi)者為同一個(gè)，即只有1個(gè)線程在消費(fèi)這9個(gè)分區(qū)的消息。

二、客戶(hù)端集成kafka

直接使用kafka客戶(hù)端，建議使用最新版的客戶(hù)端，畢竟沒(méi)有其他框架版本限制，能用最新的就用最新的，畢竟新的一般性能強(qiáng)也修復(fù)了bug。好比23年2月份出現(xiàn)的kafka安全漏洞：遠(yuǎn)程代碼執(zhí)行漏洞CVE-2023-25194，對(duì)現(xiàn)在最新版3.4.0無(wú)效，對(duì)以前大部分版本就有效。

1、添加依賴(lài)，具體如下：

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

2、發(fā)送和消費(fèi)消息，具體代碼如下：

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";
        // 創(chuàng)建消費(fèi)者
        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);
        // 訂閱主題并循環(huán)拉取消息
        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();
        // 創(chuàng)建生產(chǎn)者
        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);
        // 給主題發(fā)送消息
        producer.send(new ProducerRecord<>(topic, "hello，"+System.currentTimeMillis()));
    }
}

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

成功接收到消息并打印文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-421243.html