本篇文章为大家展示了Stream怎么在SpringCloud中使用，内容简明扼要并且容易理解，绝对能使你眼前一亮，通过这篇文章的详细介绍希望你能有所收获。
1.建一个项目，并添加如下的依赖：
<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> <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>编写下面的yml文件
server: port: 8801spring: application: name: cloud-stream-provider rabbitmq: host: 192.168.43.76 port: 5672 username: guest password: guest cloud: stream: binders: # 在此处配置要绑定的rabbitmq的服务信息； defaultRabbit: # 表示定义的名称，用于于binding整合 type: rabbit # 消息组件类型# environment: # 设置rabbitmq的相关的环境配置，（本机方式）# spring:# rabbitmq:# host: localhost# port: 5672# username: guest# password: guest bindings: # 服务的整合处理 output: # 这个名字是一个通道的名称 destination: studyExchange # 表示要使用的Exchange名称定义 content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain” binder: defaultRabbit # 设置要绑定的消息服务的具体设置(爆红不要管)eureka: client: # 客户端进行Eureka注册的配置 service-url: defaultZone: http://localhost:7001/eureka instance: lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒） lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒） instance-id: send-8801.com # 在信息列表时显示主机名称 prefer-ip-address: true # 访问的路径变为IP地址编写service，下面仅展示实现类：
import org.lzl.springcloud.service.IMessageProvider;import org.springframework.cloud.stream.annotation.EnableBinding;import org.springframework.cloud.stream.messaging.Source;import org.springframework.messaging.MessageChannel;import org.springframework.messaging.support.MessageBuilder;import javax.annotation.Resource;import java.util.UUID;//注意这里不需要写@Service，因为该service是跟rabbitmq打交道的@EnableBinding(Source.class)//定义消息的推送管道public class MessageProviderImpl implements IMessageProvider { @Resource private MessageChannel output; @Override public String send() { String serial = UUID.randomUUID().toString(); output.send(MessageBuilder.withPayload(serial).build()); System.out.println("*****serial:"+serial); return null; }}编写controller
import org.lzl.springcloud.service.IMessageProvider;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;@RestControllerpublic class SendMessageController { @Resource private IMessageProvider messageProvider; @GetMapping(value = "/sendMessage") public String sendMessage(){ return messageProvider.send(); }}案例之消息驱动消费者
写pom，加上下面的依赖
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <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-starter-actuator</artifactId> </dependency>编写yml，和生产者唯一的区别在于，下面是input
server: port: 8802spring: application: name: cloud-stream-consumer rabbitmq: host: 192.168.43.76 port: 5672 username: guest password: guest cloud: stream: binders: # 在此处配置要绑定的rabbitmq的服务信息； defaultRabbit: # 表示定义的名称，用于于binding整合 type: rabbit # 消息组件类型# environment: # 设置rabbitmq的相关的环境配置# spring:# rabbitmq:# host: localhost# port: 5672# username: guest# password: guest bindings: # 服务的整合处理 input: # 这个名字是一个通道的名称 destination: studyExchange # 表示要使用的Exchange名称定义 content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain” binder: defaultRabbit # 设置要绑定的消息服务的具体设置eureka: client: # 客户端进行Eureka注册的配置 service-url: defaultZone: http://localhost:7001/eureka instance: lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒） lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒） instance-id: receive-8802.com # 在信息列表时显示主机名称 prefer-ip-address: true # 访问的路径变为IP地址编写controller，该controller不向外界暴露端口，起到实时监控消息管道的作用！
import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Value;import org.springframework.cloud.stream.annotation.EnableBinding;import org.springframework.cloud.stream.annotation.StreamListener;import org.springframework.cloud.stream.messaging.Sink;import org.springframework.messaging.Message;import org.springframework.stereotype.Component;import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;@Component@EnableBinding(Sink.class)public class ReceiveMessageListenerController { @Value("${server.port}") private String serverPort; @StreamListener(Sink.INPUT)//只要8801发送消息，8802就会接收到8801的消息 public void input(Message<String> message){ System.out.println("消费者1号--------》接收到的消息："+message.getPayload()+"\t port: "+serverPort); }}测试
启动rabbitMQ和上面的两个项目，访问http://localhost:8801/sendMessage
在消费者的控制台中就会出现下面的订单流水号：
补充说明
我们打开rabbitmq的监控界面：发现默认是帮我们分组的
想要自定义分组只需要在消费者的yml文件中加上下面的一行：
上述内容就是Stream怎么在SpringCloud中使用，你们学到知识或技能了吗？如果还想学到更多技能或者丰富自己的知识储备，欢迎关注编程网行业资讯频道。
