springCloud Stream

Posted JoinLemon

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了springCloud Stream相关的知识,希望对你有一定的参考价值。

spring Cloud Steam

  • 生产者配置

    设置生产者的输入输出通道

package net.happyeasygo.hotel.mq.interf;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;

public interface OrderProcessor {
    String INPUT_ORDER = "inputOrder";
    String OUTPUT_ORDER = "outputOrder";

    @Input(INPUT_ORDER)
    SubscribableChannel inputOrder();

    @Output(OUTPUT_ORDER)
    MessageChannel outputOrder();
}

生产者发送消息

@EnableBinding(OrderProcessor.class)
public class PayCallBackController{
    @Autowired
    private OrderProcessor orderProcessor;
    
    public void payMethod(){
        //该data可以为对象也可以为字符串
        String data ="";
        orderProcessor.outputOrder().send(MessageBuilder.withPayload(data).build());        
       }
   }
    //添加监听消息队列output通道
    @StreamListener(OrderProcessor.INPUT_ORDER)
    public void recevieOrder(String obj) {
        _log.info("Interface order completion, message queue");
    }

设置application.yml文件

spring:
  cloud:
    stream:
      default-binder: rabbit
      bindings:
        inputOrder:
          destination: mqSupplierOrder
        outputOrder:
          destination: mqSupplierOrder
          content-type: application/json
  rabbitmq: 
    host: ${RMQ_HOST:192.168.1.57} 
    port:  ${RMQ_PORT:5672} 
    username: ${RMQ_USERNAME:guest}
    password: ${RMQ_PASSWORD:guest}
  • 消费者配置

    设置消费者的输入输出通道

package net.happyeasygo.hotel.mq.interf;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;

public interface OrderProcessor {
    String INPUT_ORDER = "inputOrder";
    String OUTPUT_ORDER = "outputOrder";

    @Input(INPUT_ORDER)
    SubscribableChannel inputOrder();

    @Output(OUTPUT_ORDER)
    MessageChannel outputOrder();
}

消费者消费消息


@EnableBinding(OrderProcessor.class)
public class OrderReceiver {

    @StreamListener(OrderProcessor.INPUT_ORDER)
    public void receiveMethod(String bookingNo) throws Exception{
        strategyDida.book("strategyDida",bookingNo);
    }
}

application.yml配置

spring:
  rabbitmq: 
    host: ${RMQ_HOST:192.168.1.57} 
    port:  ${RMQ_PORT:5672} 
    username: ${RMQ_USERNAME:guest}
    password: ${RMQ_PASSWORD:guest}
  cloud:
    stream:
      default-binder: rabbit
      bindings:
        inputOrder:
          destination: mqSupplierOrder
        outputOrder:
          destination: mqSupplierOrder

以上是关于springCloud Stream的主要内容,如果未能解决你的问题,请参考以下文章

SpringCloud——Stream(学习与使用)

#yyds干货盘点# springcloud整合stream,rabbitmq实现消息驱动功能

SpringCloud——Stream(学习与使用)

SpringCloud——Stream(学习与使用)

SpringCloud集成Stream

springcloud报错-------关于 hystrix 的异常 FallbackDefinitionException:fallback method wasn't found(代码片段