spring cloud stream集成rabbitmq

pom添加依賴

<dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>

application.yml配置

# Spring 相關 spring: # rabbitmq rabbitmq: host: 192.168.3.107 port: 5672 username: admin password: 123456

 

定義輸入通道

package com.zh.common.api.rabbitmq; import org.springframework.cloud.stream.annotation.Input; import org.springframework.messaging.SubscribableChannel; /** * @Auther: suruozhong * @Date: 2019/9/17 15:45 * @Description: */
public interface OrderChannel { //定義通道的名稱
    String saveOrder = "saveOrder"; //定義爲輸入通道
 @Input(saveOrder) SubscribableChannel saveOrder(); }

定義輸出通道

package com.zh.common.api.rabbitmq; import org.springframework.cloud.stream.annotation.Output; import org.springframework.messaging.MessageChannel; /** * @Auther: suruozhong * @Date: 2019/9/17 15:51 * @Description: */
public interface OrderOutputChannel { //定義通道的名稱
    String saveOrder = "saveOrder"; //定義爲輸入通道
 @Output(saveOrder) MessageChannel saveOrder(); }

生產端

在對應的模塊綁定輸入通道

/** * @Auther: suruozhong * @Date: 2019/7/24 15:51 * @Description: */ @SpringCloudApplication @EnableAutoConfiguration(exclude = { SecurityAutoConfiguration.class }) //經過綁定器對OderChannel通道進行綁定
@EnableBinding(OrderOutputChannel.class) public class HousekeeperFrontApplication { public static void main(String[] args) { SpringApplication.run(HousekeeperFrontApplication.class,args); } }

發送消息

@RestController @RequestMapping("/sysBanner") public class SysBannerController { @Resource private OrderOutputChannel orderOutputChannel; @RequestMapping(value = "/list") public void listData(String position) { orderOutputChannel.saveOrder().send(MessageBuilder.withPayload("fff").build()); } }

 

消費端

在對應的模塊綁定輸出通道

/** * @author * @date 2018年06月21日 * 用戶統一管理系統 */ @SpringCloudApplication @EnableScheduling //開啓定時任務 //經過綁定器對OderChannel通道進行綁定
@EnableBinding(OrderChannel.class) public class HousekeeperAdminApplication { public static void main(String[] args) { SpringApplication.run(HousekeeperAdminApplication.class, args); } }

綁定監聽消息

@Service("sysBannerService") @Transactional public class SysBannerServiceImpl extends ServiceImpl<SysBannerMapper, SysBanner> implements SysBannerService { //對saveOrder的消息監聽處理
 @StreamListener(OrderChannel.saveOrder) private void receiver(Object message){ System.out.println(message.toString()); } ｝

 

分組加持久化配置

在生產端的application.yml

spring: cloud: # spring cloud strem stream: bindings: saveOrder: 輸出通道名稱 group: saveOrder 分組名稱