kafka系列三springboot集成kafka
开始springboot项目整合kafka
1.项目依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- yml解析器 -->
<dependency>
<groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId>
</dependency>
<!--kafka-->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
2.yml文件配置
bootstrap-servers配置kafka的监听地址,集群的话配置多个地址以逗号隔开
spring:
kafka:
bootstrap-servers: 200.100.1.108:9092,200.100.1.109:9092,200.100.1.110:9092 #指定kafka server的地址,集群配多个,中间,逗号隔开
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: default_consumer_group #群组ID
#enable-auto-commit: true
#auto-commit-interval: 1000
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
server:
port: 8500
3.创建测试接口创建生产者消息
kafka0是kafka的主题,要和消费者demo文件中的topics一致
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;
/**
* @author lizhe
* @date by 2021/02/13
*/
@RestController
public class ProducerController {
@Autowired
private KafkaTemplate<String,Object> kafkaTemplate;
@RequestMapping("message/send")
public String send(String msg){
kafkaTemplate.send("kafka0", msg); //使用kafka模板发送信息
return "success";
}
}
4.创建消费者接受消息demo
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
* @author lizhe
* 监听服务器上的kafka是否有相关的消息发过来
* @date by 2021/02/13
*/
@Component
public class ConsumerDemo {
/**
* 定义此消费者接收topics = "demo"的消息,与controller中的topic对应上即可
* @param record 变量代表消息本身,可以通过ConsumerRecord<?,?>类型的record变量来打印接收的消息的各种信息
*/
@KafkaListener(topics = "kafka0")
public void listen (ConsumerRecord<?, ?> record){
System.out.printf("topic is %s, offset is %d, value is %s \n", record.topic(), record.offset(), record.value());
}
}
5.调用api接口发送消息
params加上msg参数
6.查看控制台消息
‘hellp’ 便是收到的消息
topic is kafka0, offset is 3, value is hellp
7.进入linux的kafka接受消息查看
[root@master ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.108:9092 --topic kafka0 --from-beginning
再次调用接口,会接收到消息 hellp
参考:
这篇好文章是转载于:学新通技术网
- 版权申明: 本站部分内容来自互联网,仅供学习及演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,请提供相关证据及您的身份证明,我们将在收到邮件后48小时内删除。
- 本站站名: 学新通技术网
- 本文地址: /boutique/detail/tanhffcggg
系列文章
更多
同类精品
更多
-
photoshop保存的图片太大微信发不了怎么办
PHP中文网 06-15 -
《学习通》视频自动暂停处理方法
HelloWorld317 07-05 -
Android 11 保存文件到外部存储,并分享文件
Luke 10-12 -
word里面弄一个表格后上面的标题会跑到下面怎么办
PHP中文网 06-20 -
photoshop扩展功能面板显示灰色怎么办
PHP中文网 06-14 -
微信公众号没有声音提示怎么办
PHP中文网 03-31 -
excel下划线不显示怎么办
PHP中文网 06-23 -
excel打印预览压线压字怎么办
PHP中文网 06-22 -
TikTok加速器哪个好免费的TK加速器推荐
TK小达人 10-01 -
怎样阻止微信小程序自动打开
PHP中文网 06-13