SpringCloud消息驱动 Stream

文章目录

SpringCloud消息驱动 Stream
一、SpringCloud Stream介绍
二、SpringCloud Stream常用注解简介
三、SpringCloud Stream消息驱动之生产者
四、SpringCloud Stream消息驱动之消费者
五、SpringCloud Stream重复消费
五、SpringCloud Stream持久化

一、SpringCloud Stream介绍

为什么引入cloiud Stream ？

中间件的差异性导致我们实际项目开发给我们造成了一定的困扰，我们如果用了俩个消息队列的其中一种，后面的业务需求，我想网另外一种消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都要重新推到重新做，因为它跟我们的系统耦合了，这时候SpringCloudStream给我们提供了一种解耦合的方式。
一句话：屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型。

SpringCloudStream官网

什么是SpringCloudStream？
官方定义：SpringCloudStream是一个构件消息驱动微服务的框架。
应用程序通过inputs或者outputs来与SpringCloudStream中binder对象交互。通过我们配置来binding（绑定），而SpringCloudStream的binder对象负责与消息中间件交互。所以，我们只需要搞清楚如何与SpringCloudStream交互就可以方便我们使用消息驱动的方式。
通过使用Spring Integration来连接消息代理中间件以实现消息时间驱动。
SpringCloudStream为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅，消息组，分区的三个核心概念。
目前仅支持RabbitMQ，Kafka。

Stream凭什么可以统一底层差异？
在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，他们的实现细节上会有较大的差异性，通过定义绑定器作为中间层，完美地实现了应用程序与消息中间件中之间的隔离。通过向应用程序暴露统一的Channel通道，使得应用程序不
再需要考虑各种不同的消息中间件实现。
通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离。

SpringCloudStream处理架构：

通过定义绑定器Binde作为中间层，实现了应用程序与消息中间件细节之间的隔离。

Stream中的消息通信方式遵循了发布-订阅模式。Topic主题进行广播，在RabbitMq就是Exchange，在Kafka中就是Topic。

二、SpringCloud Stream常用注解简介

Spring Cloud Strem标准流程套路：

Binder：很方便的连接中间件，屏蔽差异
Channel：通道，是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过Channel对队列进行配置。
Source和Sink：简单的可理解为参照对象是SpringCloudStream自身，从Stream发布消息就是输出，接受消息就是输入。

常用注解：

三、SpringCloud Stream消息驱动之生产者

创建maven子项目cloud-stream-rabbitmq-provider8801
Pom添加Steam支持RabbitMq的依赖

           org.springframework.cloud
            spring-cloud-starter-stream-rabbit

yml配置

server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: # 在此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关的环境配置
            spring:
              rabbitmq:
                host: 114.55.56.149
                port: 5672
                username: admin
                password: 123456
      bindings: # 服务的整合处理
        output: # 这个名字是一个通道的名称
          destination: studyExchange # 表示要使用的rabbitmq中的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-8001.com # 在信息列表时显示主机名称
    prefer-ip-address: true # 访问的路径变为ip地址

主启动类

package com.atguigu.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;

@SpringBootApplication
@EnableEurekaClient
public class StreamMQMain8801 {

    public static void main(String[] args){
        SpringApplication.run(StreamMQMain8801.class,args);
    }
}

业务类

package com.atguigu.springcloud.service;


public interface IMessageProvider {

    public String send();
}


package com.atguigu.springcloud.service.impl;

import com.atguigu.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;


@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;
    }
}

控制层

package com.atguigu.springcloud.controller;

import com.atguigu.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 {

    @Resource
    private IMessageProvider messageProvider;

    @GetMapping("/sendMessage")
    public String sendMessage(){
        return messageProvider.send();
    }

}

测试：
启动Eureka7001和provider8801,不停的访问http://localhost:8801/sendMessage.
控制台出现打印输出：

rabbitm客户端界面：

四、SpringCloud Stream消息驱动之消费者

创建子项目cloud-stream-rabbitmq-consumer8802
Pom添加依赖，同生产者一样，需要额外添加依赖


            org.springframework.cloud
            spring-cloud-starter-stream-rabbit

3.yml

server:
  port: 8802

spring:
  application:
    name: cloud-stream-consumer
  cloud:
    stream:
      binders: # 在此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关的环境配置
            spring:
              rabbitmq:
                host: 114.55.56.149
                port: 5672
                username: admin
                password: 123456
      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地址

主启动类

package com.atguigu.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;

@SpringBootApplication
@EnableEurekaClient
public class StreamMQMain8802 {

    public static void main(String[] args){
        SpringApplication.run(StreamMQMain8802.class,args);
    }
}

控制层

package com.atguigu.springcloud.controller;

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;


@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {

    @Value("${server.port}")
    private String serverPort;

    @StreamListener(Sink.INPUT)
    public void input(Message message){
        System.out.println("消费者1号，----->接收到的消息: "+ message.getPayload()+"t port: "+ serverPort);

    }
}

测试：访问五次http://localhost:8801/sendMessage后，生产者打印情况：

消费者打印情况：

五、SpringCloud Stream重复消费

我们再建一个消费者8803,这时候我们测试后发现，我们的生产者8801发出消息后，消费者8802和8803 都收到了一样的消息，存在消费重复的问题。

比如在如下场景中：订单系统我们做集群部署，都会从RabbitMq中获取订单信息，那如果一个订单同事被俩个服务获取到，那么就会造成数据错误，我们的避免这种情况。这是我们就可以使用Stream中的消息分组来解决。

注意在Stream中处于同一个group中的多个消费者是竞争关系，就能保证消息只会被其中一个应用消费一次。不同组是可以全面消费的（重复消费）。
比如我们三面的例子，来个消费者就是在俩个组中。如何知晓呢？

登陆我们的rabbitMq客户端，点击我们的studyExchange
发现有俩个Bindings,分别点击后，发现他们的流水号是不一样的。

所以我们的故障现象：重复消费
导致原因：默认分组group是不同的，组流水号不一样，被认为不同组，可以消费。
如何解决呢？
自定义配置分组
自定义配置分为同一个组，解决重复消费问题

原理：微服务应用放置于用一个group中，就能够保证消息只会被其中一个应用消费一次。不同的组是可以重复消费的，同一个组内会发生竞争关系，只有其中一个可以消费。

代码中如何实现了呢？我们只需要消费者yml中间中添加一下配置即可，表示所在组名。

登陆我们的RabbitMq客户端发现他们的名称也发生了变化。

五、SpringCloud Stream持久化

我们测试：当我们去掉消费者8802的分组配置。然后停掉消费者8802和8803的服务。而我们的生产者8801依旧在发送消息，发完消息后，我们在启动我们的消费者8802和8803，发现去掉组名配置的8802无法接收到刚刚掉线错过的消息。而我们自定义配合了组名的消费者8803依旧会接收到我们的刚刚掉线，生产者8801发送的消息！这就是SpringCloudStream的持久化。

SpringCloud消息驱动 Stream

大数据系统相关栏目本月热门文章