spring boot 集成kafka ，并且实现 发送信息，进行消费信息（亲测有效）

        <!--        kafka-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.5.4.RELEASE</version>
        </dependency>

spring:
  kafka:
    # kafka 所在IP 与 端口
    bootstrap-servers: 127.0.0.1:9092
    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
    consumer:
      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      auto-offset-reset: earliest
      # 是否自动提交offset
      enable-auto-commit: true
      # 提交offset延时(接收到消息后多久提交offset)
      auto-commit-interval: 1000
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

package com.jing.db2word.postgresql.kafka.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.tomcat.util.bcel.Const;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * 发送者
 * */
@Slf4j
@Component
public class Producer {
    private ObjectMapper objectMapper = new ObjectMapper();
    @Autowired

    private KafkaTemplate<String, Object> kafkaTemplate;


    /**
     * 封装一下 发送信息的底层逻辑，只是topic 不一样
     * @param obj  发送的具体信息
     */
    public void geojsonSync(Object obj) {
        try {
            sendMsg("jing-test", objectMapper.writeValueAsString(obj));
        } catch (JsonProcessingException e) {
            log.error("jing-test  ===> 发生异常:", e);
        }
    }


    /**
     * 底层发送信息
     * @param topic 主题
     * @param message  具体信息
     * */
    public void sendMsg(String topic, Object message) {

        if (!StringUtils.isEmpty(topic)) {
            kafkaTemplate.send(topic, message).addCallback(success -> {
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();
            }, failure -> {
                log.error("发送消息失败:" + failure.getMessage());
            });
        }
    }
}

package com.jing.db2word.postgresql.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 数据消费: 这个类里面会监听topic,自动监听
 * */
@Slf4j
@Component
public class Consumer {

    //调用预处理方法
    @KafkaListener(topics = "jing-test",groupId = "jing-test")
    public void geojsonProcess(String message) {
        try {
            log.info("接收消息成功：{}:", message);
            message = message.replace("\"", "").replace("\"", "");
            //c1DExampleFc.addGeojson(message);
            System.out.println("我监听到了:"+message);
        }
        catch (Exception e){
            log.error("jing-test  ===> 发生异常:", e.getMessage());
        }
    }
}

@Api(tags = "kafka测试")
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Autowired
    private Producer producer;
    @ApiOperation(value = "发送kafka", httpMethod = "POST")
    @PostMapping("/sendKafka")
    public AjaxResult sendKafka(String code) {
        producer.geojsonSync(code);
        return AjaxResult.success("成功");
    }
}