整合Spring Cloud Stream Binder与RabbitMQ进行消息发送与接收

2021年7月1日 36点热度 0条评论

我最新最全的文章都在 南瓜慢说 www.pkslow.com ,欢迎大家来喝茶!

1 前言

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

本文将整合Spring Cloud StreamRabbitMQ,以实现消息收发。

2 整合过程

2.1 添加依赖

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>

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

2.2 定义处理收发的方法

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

发送消息:

@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);
  };
}

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

2.3 配置属性

配置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就会发送一次消息。

2.4 运行

先启动个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。

3 总结

代码请查看:https://github.com/LarryDpk/pkslow-samples

欢迎关注微信公众号<南瓜慢说>,将持续为你更新...

多读书,多分享;多写作,多整理。