SpringCloud学习笔记（九、SpringCloud Stream）

目录：

• 什么是SpringCloud Stream
• 如何使用SpringCloud Stream
• 消息分流

什么是SpringCloud Stream：

SpringCloud Stream是一个用于构建消息驱动的微服务应用框架。它通过注入，输入、输出通道来与外界通信；因此它很容易实现消息的中转，并且在更换消息中间件的时候不需要该代码，仅需要修改配置即可。支持的消息中间件如RabbitMQ、Kafka等等。

如何使用SpringCloud Stream（以RabbitMQ为例）：

1、增加maven依赖

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-test-support</artifactId>
    <scope>test</scope>
</dependency>

2、增加properties配置

spring.application.name=stream
server.port=7070

# rabbitmq
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

# stream
spring.cloud.stream.bindings.input.destination=customer
spring.cloud.stream.bindings.output.destination=customer

3、启动类加上本工程的消息代理类型

@EnableBinding({Processor.class})

@EnableBinding分为三种类型

）org.springframework.cloud.stream.messaging.Processor：接收和发送消息

）org.springframework.cloud.stream.messaging.Source：仅支持发送消息

）org.springframework.cloud.stream.messaging.Sink：仅支持接收消息

4、加上Controller及Service

@RestController
@AllArgsConstructor
public class ProcessorController {

    private final ProcessorService processorService;

    @GetMapping("/testProcessor/{message}")
    public boolean testProcessor(@PathVariable("message") String message) {
        return processorService.send(message);
    }
}

@Service
@AllArgsConstructor
public class ProcessorService {

    private final Processor processor;

    public boolean send(String message) {
        return processor.output().send(MessageBuilder.withPayload(message).build());
    }

    public boolean subscribe(MessageHandler handler) {
        return processor.input().subscribe(handler);
    }
}

5、在任意bean下写上接收逻辑或另起一个工程（另一个工程的mq需要配成一个哦）

@StreamListener(Sink.INPUT)
public void receive(String message) {
    System.err.println("receive message: " + message);
}

然后我们启动项目，访问http://localhost:7070/testProcessor/hello，此时就会在控制台看到receive message: hello的字样。

消息分流（kafka特性）：

@GetMapping("/testMessageShunt/{type}")
public boolean testMessageShunt(@PathVariable("type") String type) {
    String header = "a".equalsIgnoreCase(type) ? "msg1" : "msg2";
    return processorService.send(type, header);
}

/**
 * RabbitMQ不支持消息分流
 */
@StreamListener(value = Sink.INPUT, condition = "headers['contentType']=='mgs1'")
public void receiveMessage1(@Payload Message<String> message) {
    System.err.println("receive message1: " + message.getPayload());
}

/**
 * RabbitMQ不支持消息分流
 */
@StreamListener(value = Sink.INPUT, condition = "headers['contentType']=='mgs2'")
public void receiveMessage2(@Payload Message<String> message) {
    System.err.println("receive message2: " + message.getPayload());
}