Stream怎么在SpringCloud中使用

本篇文章为大家展示了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中使用，你们学到知识或技能了吗？如果还想学到更多技能或者丰富自己的知识储备，欢迎关注编程网精选频道。