分享更有价值
被信任是一种快乐

SpringBoot怎么整合Kafka

文章页正文上

本文小编为大家详细介绍“SpringBoot怎么整合Kafka”,内容详细,步骤清晰,细节处理妥当,希望这篇“SpringBoot怎么整合Kafka”文章能帮助大家解决疑惑,下面跟着小编的思路慢慢深入,一起来学习新知识吧。
一、准备工作
提前说明:如果你运行出问题,请检查Kafka的版本与SpringBoot的版本是否与我文中的一致,本文中的环境已经经过测试。
Kafka服务版本为 kafka_2.11-1.1.0 (Scala), 也就是1.1.0
SpringBoot版本:1.5.10.RELEASE
提前启动zk,kafka,并且创建一个Topic
[root@Basic kafka_2.11-1.1.0]# bin/kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic test_topic

确保你的kafka能够访问,如果访问不了,需要打开外网访问。
config/server.properties
advertised.listeners=PLAINTEXT://192.168.239.128:9092

Maven 依赖

org.springframework.kafka
spring-kafka


org.apache.kafka
kafka-clients
0.11.0.0

二、项目结构
为了更加体现实际开发需求,一般生产者都是在调用某些接口的服务处理完逻辑之后然后往kafka里面扔数据,然后有一个消费者不停的监控这个Topic,然后处理数据,所以这里把生产者作为一个接口,消费者放到kafka这个目录下,注意@Component注解,不然扫描不到@KafkaListener
三、具体实现代码
SpringBoot配置文件
application.yml
spring:
kafka:
bootstrap-servers: 192.168.239.128:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: test
enable-auto-commit: true
auto-commit-interval: 1000
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

生产者
package cn.saytime.web;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 测试kafka生产者
*/
@RestController
@RequestMapping(“kafka”)
public class TestKafkaProducerController {
@Autowired
private KafkaTemplate kafkaTemplate;
@RequestMapping(“send”)
public String send(String msg){
kafkaTemplate.send(“test_topic”, msg);
return “success”;
}
}
消费者
这里的消费者会监听这个主题,有消息就会执行,不需要进行while(true)
package cn.saytime.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
* kafka消费者测试
*/
@Component
public class TestConsumer {
@KafkaListener(topics = “test_topic”)
public void listen (ConsumerRecord, ?> record) throws Exception {
System.out.printf(“topic = %s, offset = %d, value = %s n”, record.topic(), record.offset(), record.value());
}
}

项目启动类
package cn.saytime;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class TestApplication{
public static void main(String[] args) {免费云主机、域名
SpringApplication.run(TestApplication.class, args);
}
}

四、测试
运行项目,执行:http://localhost:8080/kafka/send?msg=hello
控制台输出:
topic = test_topic, offset = 19, value = hello
1
为了体现消费者不止执行一次就结束,再调用一次接口:
http://localhost:8080/kafka/send?msg=kafka
topic = test_topic, offset = 20, value = kafka
1
所以可以看到这里消费者实际上是不停的poll Topic数据的。读到这里,这篇“SpringBoot怎么整合Kafka”文章已经介绍完毕,想要掌握这篇文章的知识点还需要大家自己动手实践使用过才能领会,如果想了解更多相关内容的文章,欢迎关注云技术行业资讯频道。

相关推荐: 如何实现Data URI scheme以及实图片 base64 编码

这篇文章将为大家详细讲解有关如何实现Data URI scheme以及实图片 base64 编码,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知免费云主机、域名识有一定的了解。一、 Data URI scheme 简介Data …

文章页内容下
赞(0) 打赏
版权声明:本站采用知识共享、学习交流,不允许用于商业用途;文章由发布者自行承担一切责任,与本站无关。
文章页正文下
文章页评论上

云服务器、web空间可免费试用

宝塔面板主机、支持php,mysql等,SSL部署;安全高速企业专供99.999%稳定,另有高防主机、不限制内容等类型,具体可咨询QQ:360163164,Tel同微信:18905205712

主机选购导航云服务器试用

登录

找回密码

注册