Kafka-与SpringBoot的集成

首先在pom文件中引用kafka依赖

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

一：创建一个消息发送者，发送2条消息，发送完关闭

public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("acks", "all");props.put("retries", 0);props.put("batch.size", 16384);props.put("linger.ms", 1);props.put("buffer.memory", 33554432);props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 如果不想发送String类型的内容，也可以发送自定义的类型// 但是需要为该类型写一个org.apache.kafka.common.serialization.Deserializer<T>的实现props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 第1个string泛型表示topic的类型，基本都是string，还真没见过不是string的topic// 第2个string泛型表示发送的类型，string方便一些，如果想发送对象，则需要为该对象实现一个序列化类Producer<String, String> producer = new KafkaProducer<>(props);for (int i = 0; i < 2; i++) {Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>("你的topic", "消息内容"));}// 注意此处，似乎主线程结束了send方法就不会在发送消息，// 所以不要让主线程结束，比如你可以尝试把close方法删掉试试producer.close();}

二：创建一个spring boot项目，作为消息接收者，接到消息之后打印到控制台
2.1首先在spring boot启动类中写如下代码（为了演示，对象都直接写到启动类里）

@SpringBootApplication
public class DemoApplication {public static void main(String[] args) {SpringApplication.run(DemoApplication.class, args);}@Beanpublic KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());return factory;}public ConsumerFactory<String, String> consumerFactory() {Map<String, Object> propsMap = new HashMap<>();propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 6000);propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);// 此处是消费组名字propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, "消费组");propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");return new DefaultKafkaConsumerFactory<>(propsMap);}
}

2.2写一个消费者类，也叫做监听类，发送过来的消息都会进入该类

三.写一个消费者监听类
@Component
public class Listener {// @KafkaListener(topics = {"你的topic"},id = "consumerId")@KafkaListener(topics = { "你的topic" })public void processMessage(String content) {// 此处打印发送过来的两条消息System.out.println(content);}
}

Kafka-与SpringBoot的集成相关推荐

从现在开始学 Kafka：SpringBoot 集成 Kafka，生产者与消费者示例
从现在开始学 Kafka:SpringBoot 集成 Kafka,生产者与消费者示例前言加依赖生产者加配置生产者代码示例消费者加配置消费者监听器示例调用关于 Serializer ...
kafka的单机搭建与springboot快速集成
一.kakfa相关术语 Broker 消息中间件处理节点,一个Kafka节点就是一个broker,一个或者多个Broker可以组成一个Kafka集群 Topic Kafka根据topic对消息进行归类 ...
springboot项目集成docker
文章目录一.docker常用命令 0.拉取镜像到本地仓库 1.查看所有镜像 2.创建一个新的容器并运行,返回的是容器的ID -- CONTAINER ID: 3.查看运行中的docker实例 4.查 ...
springboot的jsp应该放在哪_在springboot中集成jsp开发
springboot就是一个升级版的spring.它可以极大的简化xml配置文件,可以采用全注解形式开发,一个字就是很牛. 在springboot想要使用jsp开发,需要集成jsp,在springbo ...
springboot+mybatis集成自定义缓存ehcache用法笔记
今天小编给大家整理了springboot+mybatis集成自定义缓存ehcache用法笔记,希望对大家能有所办帮助! 一.ehcache介绍 EhCache 是一个纯Java的进程内缓存管理框架,属 ...
Springboot/Cloud集成Sentinel 和入门实战
文章目录一.Springboot/Cloud集成Sentinel 1. spring-cloud-alibaba依赖 2. 引入 Sentinel starter 3. 配置application. ...
springboot nacos_springboot集成nacos
1.现在nacos进行配置添加 2.springboot项目集成nacos nacos上有说明,根据不同的工程进行配置,如下图集成nacos客户端包 com.alibaba.boot nacos-c ...
SpringBoot 2 集成微信扫码支付
前言该文主要是手把手教你如何在SpringBoot 中集成微信扫码支付,以及集成的过程需要注意的问题事项.另外需要感谢 vbirdbest 关于微信支付和支付宝支付相关包博客总结.因为文中很多地方参 ...
MyBatis系列之--Java 项目（非SpringBoot）集成MyBatis
MyBatis系列之--Java 项目(非SpringBoot)集成MyBatis 对MyBatis简单介绍核心接口SqlSessionFactory 实战 1. Maven创建Java项目 2. ...
springboot项目集成大数据第三方dolphinscheduler调度器
文章目录摘要项目背景功能要求功能说明 1.1用例图 1.2业务流程分析 1.3业务ER图 1.4 管理任务流程图 1.5功能设计详细说明点 1.6页面原型三.本人相关其他文章链接摘要 ①d ...

Kafka-与SpringBoot的集成

Kafka-与SpringBoot的集成相关推荐

最新文章

热门文章