Springcloud整合stream,rabbitmq实现消息驱动功能

Springcloud整合stream,rabbitmq实现消息驱动功能

springcloud整合stream,rabbitmq实现消息驱动功能

1.代码实现：

创建项目stream

添加依赖

<parent>

org.springframework.boot

spring-boot-starter-parent

2.6.2

com.cxh

stream

0.0.1-SNAPSHOT

stream

Demo project for Spring Boot

8

2021.1

2021.0.0

org.springframework.cloud

spring-cloud-dependencies

${spring-cloud-dependencies.version}

pom

import

com.alibaba.cloud

spring-cloud-alibaba-dependencies

${spring-cloud-alibaba-dependencies.version}

pom

import

org.springframework.boot

spring-boot-starter

org.springframework.boot

spring-boot-starter-test

test

org.springframework.boot

spring-boot-starter-web

org.springframework.cloud

spring-cloud-starter-stream-rabbit

监听类

@EnableBinding(Sink.class)

public class SinkReceiver {

private static Logger logger = LoggerFactory.getLogger(SinkReceiver.class);

@StreamListener(Sink.INPUT)

public void receive(String payload) {

logger.info("Received: " + payload);

}

}

2.实现效果：

启动rabbitmq, 项目stream

打开浏览器http://localhost:15672/，使用账号密码guest登录rabbitmq, 在队列中发现消息：

测试send

查看控制台消息：

com.cxh.stream.SinkReceiver              : Received: 测试send

补充

Spring Cloud Stream专门用于事件驱动的微服务系统，使用消息中间件来收发信息。使用Spring Cloud Stream可专注于业务开发，而不用花太多心思在应用与MQ之间的交互上。而且，在切换MQ后，也无须做太多的代码改动。

所以Spring Cloud Stream和RabbitMQ还可以整合实现消息的收发

整合过程

添加依赖

org.springframework.cloud

spring-cloud-stream-binder-rabbit

不同的MQ使用不同的依赖，非常容易切换。

定义处理收发的方法

队列无非就是收和发，所以我们要先定义好，怎么样发，怎么样收。

发送消息：

@Bean

public Supplier pkslowSource() {

return () -> {

String message = "pkslow.com";

log.info("Sending value: " + message);

return message;

};

}

只发送一个String，一般业务通常为Entity类。这里发送的内容也固定不变，实际业务可以通过查数据库，读文件等方式获取数据源。

接收消息：

@Bean

public Consumer pkslowSink() {

return message -> {

log.info("Received message " + message);

};

}

直接打印消息即可，项目中的逻辑可按具体业务实现。

配置属性

配置RabbitMQ：

spring:  rabbitmq:    host: localhost    port: 5672    username: pkslow    password: 123456

配置Spring Cloud Stream的相关项：

spring:  cloud:    stream:      function:        definition: pkslowSource;pkslowSink      bindings:        pkslowSource-out-0:         destination: pkslow-topic        pkslowSink-in-0:          destination: pkslow-topic      poller:        fixed-delay: 500

spring.cloud.stream.function.definition会定义处理方法，如本文的收发消息的方法；

bindings配置对应的function；destination指向MQ的主题；

这里配了一个poller，每隔500ms就会发送一次消息。

运行

先启动个RabbitMQ：

docker run \

-e RABBITMQ_DEFAULT_USER=pkslow \

-e RABBITMQ_DEFAULT_PASS=123456 \

-p 5672:5672 -p 15672:15672 \

rabbitmq:3.8-management

运行程序后，会自己创建主题、发送信息、接收信息：

运行日志如下：

可以看到每一次发/收大概是间隔了500ms，当然不可能是精确的500ms。

以上就是Springcloud整合stream,rabbitmq实现消息驱动功能的详细内容，更多关于Springcloud stream rabbitmq消息驱动的资料请关注我们其它相关文章！

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。