Springcloud整合stream,rabbitmq实现消息驱动功能
作者:灰太狼_cxh 时间:2023-07-16 10:46:08
springcloud整合stream,rabbitmq实现消息驱动功能
1.代码实现:
创建项目stream
添加依赖
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.2</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.cxh</groupId>
<artifactId>stream</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>stream</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>8</java.version>
<spring-cloud-alibaba-dependencies.version>2021.1</spring-cloud-alibaba-dependencies.version>
<spring-cloud-dependencies.version>2021.0.0</spring-cloud-dependencies.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud-dependencies.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-alibaba-dependencies</artifactId>
<version>${spring-cloud-alibaba-dependencies.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
</dependencies>
监听类
@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还可以整合实现消息的收发
整合过程
添加依赖
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
不同的MQ使用不同的依赖,非常容易切换。
定义处理收发的方法
队列无非就是收和发,所以我们要先定义好,怎么样发,怎么样收。
发送消息:
@Bean
public Supplier<String> pkslowSource() {
return () -> {
String message = "www.pkslow.com";
log.info("Sending value: " + message);
return message;
};
}
只发送一个String
,一般业务通常为Entity类。这里发送的内容也固定不变,实际业务可以通过查数据库,读文件等方式获取数据源。
接收消息:
@Bean
public Consumer<String> 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。
来源:https://blog.csdn.net/weixin_39220472/article/details/122555475
![](/images/zang.png)
![](/images/jiucuo.png)
猜你喜欢
springboot异常处理的基本规范
![](https://img.aspxhome.com/file/2023/8/59488_0s.png)
java泛型中占位符T和?有什么区别
![](https://img.aspxhome.com/file/2023/6/64746_0s.png)
Java总结篇系列:Java泛型详解
java编程中字节流转换成字符流的实现方法
基于java的opencv开发过程详解
![](https://img.aspxhome.com/file/2023/8/82428_0s.png)
Java语言实现Blowfish加密算法完整代码分享
android自定义ProgressDialog加载效果
![](https://img.aspxhome.com/file/2023/7/95007_0s.png)
Android中GPS坐标转换为高德地图坐标详解
一文梳理Java 8后的新功能
Java并发编程面试之线程池
![](https://img.aspxhome.com/file/2023/4/58704_0s.png)
C#画笔使用复合数组绘制单个矩形的方法
Android带进度条的文件上传示例(使用AsyncTask异步任务)
![](https://img.aspxhome.com/file/2023/4/72624_0s.png)
Android开发中给EditText控件添加TextWatcher监听实现对输入字数的限制(推荐)
Unity3D实现批量下载图片功能
![](https://img.aspxhome.com/file/2023/3/106503_0s.png)
WPF实现魔方小游戏
![](https://img.aspxhome.com/file/2023/4/122624_0s.jpg)
Java中定时器Timer致命缺点案例详解
![](https://img.aspxhome.com/file/2023/0/66200_0s.png)