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

1 目标

有一个spring boot 项目，现在要集成kafka ，并且要实现生产者，消费者信息；

前提是我们要有一个kafka 软件，也就是kafka 是一个软件，我们得安装成功，并且可以访问

kafka windows版本的下载安装，并且本地使用（亲测有效）

以上安装成功之后，我们可以使用软件链接一下，确保我们安装这个软件成功

在这里插入图片描述

显示绿色，就是链接本地的kafka 成功

2 实现

既然本地kafka 已经安装成功，接下来我们自己的springboot 项目要集成这个kaffa ，其实和我们项目集成redis 操作一样

第一步，加入依赖

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

第二步，yml 里面添加配置

spring:kafka:# kafka 所在IP 与 端口bootstrap-servers: 127.0.0.1:9092producer:retries: 3  # 重试次数batch-size: 16384 # 批量大小buffer-memory: 33554432 # 生产端缓冲区大小key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:# earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费# latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据# none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常auto-offset-reset: earliest# 是否自动提交offsetenable-auto-commit: true# 提交offset延时(接收到消息后多久提交offset)auto-commit-interval: 1000key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

第三步，写生产者与消费者

在这里插入图片描述
以后我们的controller 或者 service 就调用生产者，消费者写好之后就自动监听信息，并且进行处理信息了，也就是把我们的业务逻辑写到消费者里面就可以

生产者里面的代码

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();@Autowiredprivate 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();// 消息在分区内的offsetlong 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());}}
}

第五步，写controller

@Api(tags = "kafka测试")
@RestController
@RequestMapping("/kafka")
public class KafkaController {@Autowiredprivate Producer producer;@ApiOperation(value = "发送kafka", httpMethod = "POST")@PostMapping("/sendKafka")public AjaxResult sendKafka(String code) {producer.geojsonSync(code);return AjaxResult.success("成功");}
}