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