您好,登錄后才能下訂單哦!
這篇文章主要介紹了spring-cloud-stream結合kafka的使用方法是什么,具有一定借鑒價值,需要的朋友可以參考下。希望大家閱讀完這篇文章后大有收獲。下面讓小編帶著大家一起了解一下。
1.pom文件導入依賴
<!-- kafka --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-kafka</artifactId> </dependency>
2.application.yml文件配置
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.創建消息發送者
@EnableBinding(Source.class) // @EnableBinding 是綁定通道的,Soure.class是spring 提供的,表示這是一個可綁定的發布通道 @Service public 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.創建消息監聽者
@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的使用方法是什么內容對大家有幫助,同時也希望大家多多支持億速云,關注億速云行業資訊頻道,遇到問題就找億速云,詳細的解決方法等著你來學習!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。