Spring Cloud系列--Stream&Binder

概要

主要介绍Spring Cloud Stream 以及 Spring Cloud Binder，一般这两个组件都是成对出现的。就是为了方便我们开发者重复造轮子。

Spring Cloud Stream

编程模型

A Spring Cloud Stream application consists of a middleware-neutral core. The application communicates with the outside world through input and output channels injected into it by Spring Cloud Stream. Channels are connected to external brokers through middleware-specific Binder implementations.

一个Spring Cloud Stream 应用主要由一个中间件组成。应用程序通过他提供的输入和输出通道来与外界通信。通过特定的中间件binder来实现通道与外部的连接。

Stream.jpg

Spring Cloud Binder

Spring Cloud Stream provides a Binder abstraction for use in connecting to physical destinations at the external middleware. A producer is any component that sends messages to a channel. The channel can be bound to an external message broker with a Binder implementation for that broker.A consumer is any component that receives messages from a channel. As with a producer, the consumer’s channel can be bound to an external message broker.

Spring Cloud Stream 提供了一种抽象的Binder用来物理连接外部的中间件。生产者可以是任意向通道发送信息的组件，通道可以使用Binder被绑定到外部的消息中间件。消费者也可以通过任意组件从通道中获取消息。与生产者一样，消费者的通道也可以被绑定到外部消息中间件。

binder.jpg

简单使用Kafka实现消息通信

下面就让我们简单的实现一个消息的推送和订阅。大概架构流程图 如下

流程.jpg

基于前一篇组件的基础上。增加Kafka中间件即可。

实现服务方（接收端） --demo-provider

目录

provider目录.jpg

增加pom.xml

<!-- 依赖 Spring Cloud Stream Binder Kafka -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>

yml文件修改

server:
  port: 8088
spring:
  application:
    name: demo-provider
  cloud:
    stream:
      bindings:
        input:
          destination: users        # 监听的topic
          contentType: text/plain   # 传送类型（选填）
      kafka:
        binder:
          brokers: xx.xxx.xx.xx   #Kafka 地址

eureka:
  client:
    serviceUrl:
      defaultZone: http://localhost:1111/eureka/

添加 Stream 绑定接口

public interface UserMessage {

    String INPUT = "input";

    //定义输出源
    // 这里就是Spring Cloud Stream提供的输入通道
    @Input(INPUT)
    SubscribableChannel input();
}

添加通道监听

@Service
public class UserMessageListener {


    //Spring Cloud Stream 提供的监听接口
    //监听定义好的输出通道
    @StreamListener(UserMessage.INPUT)
    public void streamListenerOnMessage(String name) {
        System.out.println("Stream on  @StreamListener");
        System.out.println(name);
    }
}

激活绑定

@EnableDiscoveryClient  //服务客户端注册
@SpringBootApplication
@EnableBinding(UserMessage.class)  //绑定Spring Cloud Stream 定义好的通道接口
public class DemoApplication {

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

}

实现消费端（发送方）--demo-consumer

目录

consumer目录.jpg

修改pom.xml

<!-- 依赖 Spring Cloud Stream Binder Kafka -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>

修改yml文件

server:
  port: 8082
spring:
  application:
    name: demo-consumer
  cloud:
    stream:
      bindings:
        output:
          destination: users     #发送的topic 
          contentType: text/plain  #文本类型 
      kafka:
        binder:
          brokers: xx.xxx.xx.xx  # Kafka 服务器地址
eureka:
  client:
    serviceUrl:
      defaultZone: http://localhost:1111/eureka/

添加Stream绑定接口

public interface UserMessage {

    String OUTPUT = "output";
    //定义输入源
    @Output(OUTPUT)
    MessageChannel output();
}

添加消息推送方法

@RestController
public class UserMessageController {

    //装配接口通道
    @Autowired
    private UserMessage userMessage;

    
    @GetMapping("/send/name")
    public boolean sendUserMessage(@RequestParam("name") String name) {
        //获取通道
        MessageChannel messageChannel = userMessage.output();
        //格式化消息
        GenericMessage<String> message = new GenericMessage<String>(name);
        System.out.println("begin to send ......" + name);
        //发送
        return messageChannel.send(message);
    }
}

激活绑定

@SpringBootApplication
@EnableFeignClients(clients = HelloWorldService.class)   //申明HelloWorldService作为Feign的调用
@EnableBinding(UserMessage.class)
public class DemoApplication {

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

}

运行以及结果

启动Kafka

这里需要大家安装Kafka中间件，这里就不多做介绍了。可以参照其他大神的博客来安装启动。

启动其他组件

这里注意启动顺序

1. demo-register:上一篇有demo，可以copy过来启动即可。
2. demo-provider
3. demo-consumer

除了注册中心，其他俩个服务，出现类似如下图，即连接Kafka成功.

kafka-spring.jpg

查看Kafka Topic

topic.jpg

我们需要查看Kafka中是否有我们在 Spring中定义的 Topic ： users。

我们需要登录到kafka机器，进入到Kafka目录，输入如下命令查看

bin/kafka-topics.sh --list --zookeeper localhost:2181

如果出现如下内容，则代表启动没有问题，可以进行请求测试了。

请求以及查看发送和监听内容。

使用postman GET方法 请求路径 http://localhost:8082/send/name?name=neal

服务发送方

send.jpg

服务接收方

accept.jpg

我们发现无论是监听和发送都运行正常。这里简单的实现就介绍完了。

小结

Spring Stream 相关组件，在有消息推送的项目和业务中，可以起到非常快速开发的作用，但是也需要我们多了解才可以运用自如。