SpringBoot之kafka对接topic

第一步，引入pom文件

     <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

第二步：引入配置文件

spring:kafka:bootstrap-servers: 192.168.227.141:9092producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:group-id: testenable-auto-commit: trueauto-commit-interval: 1000key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

第三步：注入配置文件

package com.my.tool;import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.util.backoff.FixedBackOff;@Component
public class KafkaConfig {@Beanpublic ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(ConcurrentKafkaListenerContainerFactoryConfigurer configurer,ConsumerFactory<Object, Object> kafkaConsumerFactory,KafkaTemplate<Object, Object> template) {ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();configurer.configure(factory, kafkaConsumerFactory);factory.setErrorHandler(new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(template), new FixedBackOff(0L, 2))); // dead-letter after 3 triesreturn factory;}}

第四步：发送消息

package com.my.tool;import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;@Component
public class KafkaSender {private final org.slf4j.Logger log = LoggerFactory.getLogger(getClass());@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;//发送消息方法public void send(String strjson) {log.info("+++++++++++++++++++++  message = {}", strjson);kafkaTemplate.send("test", strjson);}
}

第五步：接收消息

package com.my.tool;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;import java.util.Optional;@Component
public class KafkaReceiver {private final org.slf4j.Logger log = LoggerFactory.getLogger(getClass());@KafkaListener(topics = {"test"})public void listen(ConsumerRecord<?, ?> record) {Optional<?> kafkaMessage = Optional.ofNullable(record.value());if (kafkaMessage.isPresent()) {Object message = kafkaMessage.get();log.info("----------------- record =" + record);log.info("------------------ message =" + message);}}
}

第六步：写一个controller测试

package com.my.controller;import com.my.tool.KafkaSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;import java.io.IOException;@RestController
public class PorduceController {@AutowiredKafkaSender kafkaSender;@GetMapping("/Porduce")public void Produce(String Message) throws IOException {kafkaSender.send(Message);}
}

SpringBoot之kafka对接topic相关推荐

SpringBoot操作Kafka创建Topic、Producer、Consumer
环境 kafka 2.6.0(安装步骤查看这里) 引入依赖 <dependency><groupId>org.springframework.boot</groupId& ...
springboot集成kafka及kafka web UI的使用
springboot集成kafka application.properties spring.kafka.bootstrap-servers=CentOSA:9092,CentOSB:9092,Ce ...
SpringBoot整合kafka(实现producer和consumer)
转载自 SpringBoot整合kafka(实现producer和consumer) 在Windows环境下安装运行Kafka:https://www.jianshu.com/p/d64798e81f ...
springboot 整合 kafka demo 顺便看一下源码
大家好,我是烤鸭: 今天分享下 springboot 整合 kafka. 1. 环境参数: windows + kafka_2.11-2.3.0 + zookeeper-3.5.6 + spring ...
SpringBoot整合kafka之kafka分区实战
本文来说下SpringBoot整合kafka之kafka分区实战文章目录准备工作程序代码程序测试本文小结准备工作当然我们也可以不手动创建topic,在执行代码kafkaTemplate. ...
SpringBoot整合kafka实战之带回调的生产者
本文来说下SpringBoot整合kafka部分知识内容文章目录带回调的生产者方式一方式二本文小结带回调的生产者前面我们说了简单的生产和消费,本文说下带回调的生产者.kafkaTempl ...
SpringBoot 重置 kafka 偏移量offset（kafka-0.10.1.0）
如果使用的自动提交偏移量的模式,偏移量会给到kafka或者zk进行管理,其中kafka的偏移量重置给了重新消费kafka内未过期的数据提供了机会,当消费者出错,比如消费了数据,但是中途处理失败,导致数 ...
从现在开始学 Kafka：SpringBoot 集成 Kafka，生产者与消费者示例
从现在开始学 Kafka:SpringBoot 集成 Kafka,生产者与消费者示例前言加依赖生产者加配置生产者代码示例消费者加配置消费者监听器示例调用关于 Serializer ...
Springboot与Kafka的小插曲
目录 1.背景 2.环境 3.应用 1)pom.xml 2)application.yml 3)main方法 4)Entity 5)ServiceImpl 6)controller 4.遇到的问题 5 ...

SpringBoot之kafka对接topic

SpringBoot之kafka对接topic相关推荐

最新文章

热门文章