标签：报错 bubuko ret org row listen deb actor 接收
1、POM引入spring-cloud-starter-stream-rabbit
注意springboot 版本，用1.5.2.RELEASE 就报错 rabbitmqTemple not found
sb版本切换到1.5.14 就好了
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.14.RELEASE</version>
<relativePath /> <!-- lookup parent from repository -->
</parent>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>Edgware.SR4</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
public interface OrderStream { @Output(ORDER_OUT) MessageChannel outOrder(); @Input(ORDER_INPUT) SubscribableChannel inputOrder(); @Output(ORDER_OUT_R) MessageChannel outOrderR(); @Input(ORDER_INPUT_R) SubscribableChannel inputOrderR(); }
ORDER_OUT 用用于发送消息，ORderOUT_R 用于接收消息回调
如果input 里面的名称 和 output 配置一样会报错
3、发送
指定bindings
@SuppressWarnings("SpringJavaAutowiringInspection")
@EnableBinding({OrderStream.class})
public class OrderMsgProceducer {
private static final Logger log = LoggerFactory.getLogger(OrderMsgProceducer.class);
@Autowired
@Qualifier(OrderStream.ORDER_OUT)
private MessageChannel messageChannel;
public void send(Order timingEvent) {
Message<Order> message = MessageBuilder.withPayload(timingEvent).build();
messageChannel.send(message);
log.info("send Mq:{}",message);
}
}
4、接受 与回调
之所以能够接受到是因为
接受的与发送的是一个
@SuppressWarnings("SpringJavaAutowiringInspection")
@EnableBinding({OrderStream.class})
public class OrderMsgConsumer {
private final static Logger log = LoggerFactory.getLogger(OrderMsgConsumer.class);
@StreamListener(OrderStream.ORDER_OUT)
@SendTo(OrderStream.ORDER_OUT_R)
public String receiver(@Payload Order order) throws Exception {
log.info("Receive Order To MQ:{}",order);
log.debug("MQ END:{}",order);
return "我收到了";
}
@StreamListener(OrderStream.ORDER_OUT_R)
public void receiverHuidiao(String msg) throws Exception {
System.out.println("我接收回调......");
System.out.println(msg);
}
}
配置文件
spring.rabbitmq.host=1xxx spring.rabbitmq.port=5673 spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.cloud.stream.bindings.ORDER_SEND.content-type=application/json;charset=UTF-8 spring.cloud.stream.bindings.ORDER_SEND.producer.deliveryMode=PERSISTENT spring.cloud.stream.bindings.ORDER_SEND.destination=ORDER_14 spring.cloud.stream.bindings.ORDER_SEND.group=becom spring.cloud.stream.bindings.ORDER_SEND.consumer.maxAttempts=3 spring.cloud.stream.bindings.ORDER_SEND.consumer.backOffInitialInterval=1000 spring.cloud.stream.bindings.ORDER_SEND.consumer.concurrency=1
group作用：如果我们启动两个服务，能够确保我们发送一个消息只被消费一次
测试
mq页面
项目：
SpringCloud stream连接RabbitMQ收发信息
标签：报错 bubuko ret org row listen deb actor 接收
原文地址：https://www.cnblogs.com/lyon91/p/10118383.html
