配置 RabbitMQ 消息队列和 Stream 消息发送与接收的通道。
server: port: 8001 # 端口 spring: application: name: stream-producer # 应用名称 rabbitmq: host: 192.168.10.101 # 服务器 IP port: 5672 # 服务器端口 username: guest # 用户名 password: guest # 密码 virtual-host: / # 虚拟主机地址 cloud: stream: bindings: # 消息发送通道 # 与 org.springframework.cloud.stream.messaging.Source 中的 @Output("output") 注解的 value 相同 output: destination: stream.message # 绑定的交换机名称 # 配置 Eureka Server 注册中心 eureka: instance: prefer-ip-address: true # 是否使用 ip 地址注册 instance-id: ${spring.cloud.client.ip-address}:${server.port} # ip:port client: service-url: # 设置服务注册中心地址 defaultZone: :8761/eureka/,:8762/eureka/发送消息
MessageProducer.java
package com.example.producer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.messaging.Source; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Component; /** * 消息生产者 */ @Component @EnableBinding(Source.class) public class MessageProducer { @Autowired private Source source; /** * 发送消息 * * @param message */ public void send(String message) { source.output().send(MessageBuilder.withPayload(message).build()); } }启动类
StreamProducerApplication.java
package com.example; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class StreamProducerApplication { public static void main(String[] args) { SpringApplication.run(StreamProducerApplication.class); } }消息消费者
创建项目
在 stream-demo 项目下创建 stream-consumer 子项目。
添加依赖
要使用 RabbitMQ 绑定器，可以通过使用以下 Maven 坐标将其添加到 Spring Cloud Stream 应用程序中：
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-rabbit</artifactId> </dependency>或者使用 Spring Cloud Stream RabbitMQ Starter：
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId> </dependency>
完整依赖如下：
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 "> <modelVersion>4.0.0</modelVersion> <groupId>com.example</groupId> <artifactId>stream-consumer</artifactId> <version>1.0-SNAPSHOT</version> <!-- 继承父依赖 --> <parent> <groupId>com.example</groupId> <artifactId>stream-demo</artifactId> <version>1.0-SNAPSHOT</version> </parent> <!-- 项目依赖 --> <dependencies> <!-- netflix eureka client 依赖 --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId> </dependency> <!-- spring cloud stream binder rabbit 绑定器依赖 --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-rabbit</artifactId> </dependency> <!-- spring boot test 依赖 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <exclusions> <exclusion> <groupId>org.junit.vintage</groupId> <artifactId>junit-vintage-engine</artifactId> </exclusion> </exclusions> </dependency> </dependencies> </project>配置文件