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

更新时间：2022年02月25日 16:02:48 作者：灰太狼_cxh

官方定义SpringCloud Stream 是一个构建消息驱动微服务的框架。我们只需要搞清楚如何与Spring Cloud Stream 交互就可以方便使用消息驱动的方式。本文将通过Springcloud整合stream,rabbitmq实现消息驱动功能，需要的可以参考一下

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。

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

您可能感兴趣的文章:

springboot oauth2实现单点登录实例
我们见过的很多网站，容许使用第三方账号登录，oauth2是用来做三方登录的，本文就详细的介绍springboot oauth2实现单点登录实例，具有一定的参考价值，感兴趣的可以了解一下
2022-01-01
Java面试突击之Bean作用域详解
Spring 框架作为一个管理 Bean 的 IoC 容器，那么 Bean 自然是 Spring 中的重要资源了，那 Bean 的作用域是什么意思？又有几种类型呢？接下来我们一起来看
2022-08-08
Java Scanner的使用和hasNextXXX()的用法说明
这篇文章主要介绍了Java Scanner的使用和hasNextXXX()的用法说明，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-10-10
java实现简单猜拳小游戏
这篇文章主要为大家详细介绍了java实现简单猜拳小游戏，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2021-11-11
Java设计模式之代理模式原理及实现代码分享
这篇文章主要介绍了Java设计模式之代理模式原理及实现代码分享，设计代理模式的定义，静态代理，动态代理，jdk动态代理实现步骤，原理及源码等相关内容，具有一定参考价值，需要的朋友可以了解下。
2017-11-11
Java多线程Queue、BlockingQueue和使用BlockingQueue实现生产消费者模型方法解析
这篇文章主要介绍了Java多线程Queue、BlockingQueue和使用BlockingQueue实现生产消费者模型方法解析，涉及queue，BlockingQueue等有关内容，具有一定参考价值，需要的朋友可以参考。
2017-11-11
Spring IOC容器Bean管理XML注入集合类型属性
这篇文章主要为大家介绍了Spring IOC容器Bean管理XML注入集合类型属性，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2022-05-05
hystrix配置中Apollo与Archaius对比分析
这篇文章主要为大家介绍了hystrix的配置中Apollo与Archaius对比分析，并为大家解答在hystrix的配置中有了Apollo是否还需要Archaius这一问题详解
2022-02-02
IntellJ IDEA神器使用技巧(小结)
这篇文章主要介绍了IntellJ IDEA神器使用技巧(小结)，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-06-06
javax.servlet不存在的解决办法
今天在关于servlet的编译中出现了一下小小的错误，发现javax.servlet不存在，本文就介绍一下解决方法，感兴趣的可以了解一下
2021-07-07

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

相关文章

最新评论

大家感兴趣的内容

最近更新的内容

常用在线小工具