0


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.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
      # 是否自动提交offsetenable-auto-commit:true# 提交offset延时(接收到消息后多久提交offset)auto-commit-interval:1000key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

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

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

生产者里面的代码

packagecom.jing.db2word.postgresql.kafka.producer;importcom.fasterxml.jackson.core.JsonProcessingException;importcom.fasterxml.jackson.databind.ObjectMapper;importlombok.extern.slf4j.Slf4j;importorg.apache.commons.lang3.StringUtils;importorg.apache.tomcat.util.bcel.Const;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.stereotype.Component;/**
 * 发送者
 * */@Slf4j@ComponentpublicclassProducer{privateObjectMapper objectMapper =newObjectMapper();@AutowiredprivateKafkaTemplate<String,Object> kafkaTemplate;/**
     * 封装一下 发送信息的底层逻辑,只是topic 不一样
     * @param obj  发送的具体信息
     */publicvoidgeojsonSync(Object obj){try{sendMsg("jing-test", objectMapper.writeValueAsString(obj));}catch(JsonProcessingException e){
            log.error("jing-test  ===> 发生异常:", e);}}/**
     * 底层发送信息
     * @param topic 主题
     * @param message  具体信息
     * */publicvoidsendMsg(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());});}}}

消费者里面的代码

packagecom.jing.db2word.postgresql.kafka.consumer;importlombok.extern.slf4j.Slf4j;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;/**
 * 数据消费: 这个类里面会监听topic,自动监听
 * */@Slf4j@ComponentpublicclassConsumer{//调用预处理方法@KafkaListener(topics ="jing-test",groupId ="jing-test")publicvoidgeojsonProcess(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")publicclassKafkaController{@AutowiredprivateProducer producer;@ApiOperation(value ="发送kafka", httpMethod ="POST")@PostMapping("/sendKafka")publicAjaxResultsendKafka(String code){
        producer.geojsonSync(code);returnAjaxResult.success("成功");}}

以上就写好了,我们进行测试

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
发现控制台也输出流,kafka 软件里面也有记录,所以我们项目就集成了


本文转载自: https://blog.csdn.net/python113/article/details/138534237
版权归原作者 一写代码就开心 所有, 如有侵权,请联系我们删除。

“spring boot 集成kafka ,并且实现 发送信息,进行消费信息(亲测有效)”的评论:

还没有评论