当前位置：首页 > news >正文

SpringCloud系列（49）--SpringCloud Stream消息驱动之实现生产者

news 2025/7/4 8:19:05

前言：在上一节中我们简单的介绍了SpringCloud Stream的相关知识，而这一节则是关于如何去实现SpringCloud Stream消息驱动里的生产者

注：在实现SpringCloud Stream生产者前需要电脑具备RabbitMQ环境，没有具备的需要先行安装

下述的教程在官方操作手册里有说到一部分，有兴趣的同学可以参考下

官方操作手册：

https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/

进入Rabbit MQ Binder的教程（本节教程用的消息队列是Rabbit）

1、创建一个生产者模块，命为cloud-stream-rabbitmq-provider8801

(1)在父工程下新建模块

(2)选择模块的项目类型为Maven并选择模块要使用的JDK版本

(3)填写子模块的名称，然后点完成即可完成创建

效果图：

(4)修改cloud-stream-rabbitmq-provider8801子模块的pom.xml文件，然后reolad一下，下载依赖

例：

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>springcloud01</artifactId><groupId>com.ken.springcloud</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>cloud-stream-rabbitmq-provider8801</artifactId><dependencies><!--以下依赖都没写版本号，没写版本号的情况下会引用父项目的版本--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!--监控--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><!--Eureka Clinet--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><!--热部署--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><!--lombok插件--><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies></project>

(5)为子模块添加名为application的yml配置文件

效果图：

(6)修改application.yml文件配置（注：因为当前实现的是生产者，所以spring.cloud.stream.bindings下要写成output，表示当前是生产者）

server:port: 8801spring:application:name: cloud-stream-providercloud:stream:#在此处配置要绑定的rabbitmq的服务信息binders:#表示定义的名称，用于于binding整合defaultRabbit:#消息组件类型，表示用到的组件是rabbittype: rabbit#设置rabbitmq的相关的环境配置environment:spring:rabbitmq:host: localhostport: 5672username: guestpassword: guest#服务的整合处理bindings:#这个名字是一个通道的名称output:#表示要使用的Exchange名称定义destination: studyExchange#设置消息类型，本次为json，文本则设置"text/plain"content-type: application/json#设置要绑定的消息服务的具体设置binder: defaultRabbiteureka:#客户端进行Eureka注册的配置client:service-url:defaultZone: http://localhost:7001/eurekainstance:#设置心跳的时间间隔（默认是30秒)lease-renewal-interval-in-seconds: 2#如果现在超过了5秒的间隔（默认是90秒)lease-expiration-duration-in-seconds: 5#在信息列表时显示主机名称instance-id: send-8001.com#访问的路径变为IP地址prefer-ip-address: true

(7)为cloud-stream-rabbitmq-provider8801子模块新建一个主启动类，类名输入com.ken.springcloud.StreamMQMain8801，然后创建即可

效果图：

(8)编写StreamMQMain8801主启动类

package com.ken.springcloud;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class StreamMQMain8801 {public static void main(String[] args) {SpringApplication.run(StreamMQMain8801.class, args);}
}

2、在com.ken.springcloud包下新建一个名为service的包

效果图：

3、在service包下新建一个名为IMessageProvider的接口类

效果图：

4、编写IMessageProvider接口类

package com.ken.springcloud.service;public interface IMessageProvider {public String send();}

5、在service包下新建一个名为impl的包

效果图：

6、在service.impl包下新建一个名为IMessageProviderImpl的实现类

效果图：

7、编写IMessageProviderImpl类

package com.ken.springcloud.service.impl;import com.ken.springcloud.service.IMessageProvider;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;import javax.annotation.Resource;
import java.util.UUID;//@EnableBinding(Source.class)用于定义消息的推送管道
@EnableBinding(Source.class)
public class IMessageProviderImpl implements IMessageProvider {@Resource//消息发送管道private MessageChannel output;@Overridepublic String send() {String serial = UUID.randomUUID().toString();output.send(MessageBuilder.withPayload(serial).build());System.out.println("serial:" + serial);return null;}
}

8、在com.ken.springcloud包下新建一个名为controller的包

效果图：

9、在controller包下新建一个名为SendMessageController的控制类

效果图：

10、编写SendMessageController类

package com.ken.springcloud.controller;import com.ken.springcloud.service.IMessageProvider;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;@RestController
public class SendMessageController {@Resourceprivate IMessageProvider iMessageProvider;@GetMapping("/sendMessage")public String sendMessage() {return iMessageProvider.send();}}