SpringCloud Stream 整合RabbitMQ的基本步骤_Java教程

SpringCloud Stream 整合RabbitMQ的基本步骤

2022-10-28 11:55毛毛的猫毛 Java教程

这篇文章主要介绍了SpringCloud Stream 整合RabbitMQ的基本步骤，从项目介绍到生产者结合示例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下

本篇简单介绍SpringCloud Stream 整合RabbitMQ基本步骤：

引入SpringCloud
引入SpringCloud Stream相关依赖
定义绑定接口: 消息生产者(Output…Binding) 、消息消费者(Input…Binding)
@EnableBinding 在对应类上进行定义
@StreamListener 在对应方法上创建监听用来消费消息
调用output的send()方法生产消息

一、项目介绍

演示SpringCloud Stream 整合RabbitMQ,项目可以在一个工程里完成，本次建立了一个工程mq-service,其中包含三个Module:

mq-service-base ：基础模块(包含了共用依赖、共用变量)
mq-service-producer ：生产者
mq-service-consumer ：消费者

注：完全可以在一个工程里实现，这里为了区分，并为了后续单独启动或停止生产者或消费者做实验，也为了适应实际应用项目，所以创建了不同Module

（1）版本

SpringBoot : 2.0.6.RELEASE
SpringCloud : Finchley.SR2
RabbitMQ : 3.8.1

（2）项目整体结构

SpringCloud Stream 整合RabbitMQ的基本步骤

（3）基础模块

1）pom.xml

这里作为公共模块引入SpringCloud、Spring Cloud Stream等，其中也再此引入fastjson、lombok等工具依赖
（完整代码见文章最下面）
其中Spring Cloud Stream如下：

<!-- Spring Cloud Stream, 用于MQ消息发送-->
<dependency>
	<groupId>org.springframework.cloud</groupId>
	<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
	<groupId>org.springframework.cloud</groupId>
	<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

2) model

定义共用的变量，如CollectionRequest.java

二、生产者

（1）结构

SpringCloud Stream 整合RabbitMQ的基本步骤

（2）pom.xml

导入base的依赖即可，因为相关共用依赖在base中已经引入

<dependency>
	<groupId>com.zrk</groupId>
	<artifactId>mq-service-base</artifactId>
	<version>0.0.1-SNAPSHOT</version>
</dependency>

（3）定义绑定(接口)

OutputMessageBinding.java

public interface OutputMessageBinding {
  /** Topic 名称*/
  String OUTPUT = "message-center-out";
  @Output(OUTPUT)
  MessageChannel output();

}

（4）添加配置

# rabbitmq连接信息
spring.rabbitmq.addresses=192.168.1.125
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

spring.cloud.stream.bindings.message-center-out.destination=message-center
spring.cloud.stream.rabbit.bindings.message-center-out.consumer.exchangeType=fanout

(5) 调用方法

CollectionServiceImpl.java

@Service
@EnableBinding(OutputMessageBinding.class)
public class CollectionServiceImpl implements CollectionService{
  @Resource
  private OutputMessageBinding outputMessageBinding;
  /**
   * @param schoolName
   * @param content
   */
  @Override
  public void getCollection(String schoolName, String content) {
      CollectionRequest request = new CollectionRequest();
      request.setSchoolName(schoolName);
      request.setContent(content);
      outputMessageBinding.output().send(MessageBuilder.withPayload(request).build());
  }
}

注：主要是两点

@EnableBinding 定义
outputMessageBinding.output().send(MessageBuilder.withPayload(request).build()); 生产消息

三、消费者

（1）结构

SpringCloud Stream 整合RabbitMQ的基本步骤

（2）pom.xml

导入base的依赖即可，因为相关共用依赖在base中已经引入

<dependency>
	<groupId>com.zrk</groupId>
	<artifactId>mq-service-base</artifactId>
	<version>0.0.1-SNAPSHOT</version>
</dependency>

（3）定义绑定(接口)

InputMessageBinding.java

public interface InputMessageBinding {
  String INPUT = "message-center-input";
  @Input(INPUT)
  SubscribableChannel input();
}

注：消费者这里与生产者不同，用的是SubscribableChannel ，而生产者用的是MessageChannel

（4）添加配置

# rabbitmq连接信息
spring.rabbitmq.addresses=192.168.1.125
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456
spring.cloud.stream.bindings.message-center-input.destination=message-center
spring.cloud.stream.bindings.message-center-input.group=${spring.application.name}

(5) 调用方法

CollectionReceiver.java

@Slf4j
@EnableBinding(InputMessageBinding.class)
public class CollectionReceiver {
  @StreamListener(InputMessageBinding.INPUT)
  public void handle(String value){
      log.info("[消息] 接收到发送消息MQ: {}", value);
      CollectionRequest request = JSON.parseObject(value, CollectionRequest.class);
      log.info("处理收集信息：" + request.toString());
  }
}

注：主要是两点