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  分組名稱