1.pom文件導入依賴
2.application.yml文件配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
spring: cloud: stream: kafka: binder: brokers: xxx.xxx.xxx.xx:xxxx // Kafka的消息中間件服務器地址 bindings: xxx_output: // 通道名稱 destination: xxx // 消息發往的目的地，對應topic 在發送消息的配置里面，group是不用配置的 // 如果我們需要傳輸json的信息，那么在發送消息端需要設置content-type為json（其實可以不寫，默認content-type就是json） xxx_input: destination: xxx // 消息發往的目的地，對應topic group: xxx // 對應kafka的group |
3.創建消息發送者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
@EnableBinding(Source.class) // @EnableBinding 是綁定通道的，Soure.class是spring 提供的，表示這是一個可綁定的發布通道@Servicepublic class MqService { @Resource(name = KafkaConstants.OES_WORKBENCH_LIFE_DATA_OUTPUT) private MessageChannel oesWorkbenchChannel; /** * 發送一條kafka消息 */ public boolean sendLifeData(Object object) { return MqUtils.send(oesWorkbenchChannel, object, KafkaConstants.OES_WORKBENCH_LIFE_DATA_OUTPUT); }}// 發布通道public interface Source { @Output(KafkaConstants.OES_WORKBENCH_LIFE_DATA_OUTPUT) MessageChannel oesWorkbenchLifeDataOutput(); // 發布通道用MessageChannel } |
4.創建消息監聽者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
@Slf4j@EnableBinding(Sink.class)public class WorkbenchStreamListener { @Resource private FileService fileService; @StreamListener(KafkaConstants.xxx_input) // 監聽接受通道 public void receiveData(MoveMessage moveMessage) { }}// 接受通道public interface Sink { @Input(KafkaConstants.OES_WORKBENCH_MOVE_INPUT) SubscribableChannel oesWorkbenchMoveInput(); // 接受通道用SubscribableChannel } |
接下來就可以愉快的發送監聽消息了
到此這篇關于spring-cloud-stream結合kafka使用詳解的文章就介紹到這了,更多相關spring-cloud-stream整合kafka內容請搜索服務器之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持服務器之家！
原文鏈接：https://blog.csdn.net/oWanShiKaiTouNan/article/details/108056417
