SpringCloud Stream操作消息队列_springcloud stream 仲裁队列

概述

SpringCloud Stream是为微服务提供消息驱动能力的组件，使用就可以操作消息队列。

应用程序通过input和output与stream的binder通信，而binder与消息中间件(MQ)通信。使用stream是对消息中间件的进一步封装，无视不同中间件的API，做到统一封装，可以做到代码层面的无感知，甚至动态切换中间件，不过只目前只支持RabbitMQ和kafka。

集成
pom引入依赖，我使用的是rabbitmq，使用kafka就引入kafka

<dependency>
   <groupId>org.springframework.cloud</groupId>
   <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
1
2
3
4

配置文件配置mq信息，将消息分组，queue为myMessageInput.order，消息格式为json

spring:
  rabbitmq:
    host: 
    port: 5672
    username: guest
    password: guest
  cloud:
    stream:
      bindings:
        myMessageInput:
          group: order
          content-type: application/json
1
2
3
4
5
6
7
8
9
10
11
12

新建一个Stream接口 StreamClient

package com.viki.order.message;
 
 
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
 
/**
 * @Author Sakura
 * @Date 10/11/2019
 **/
 
public interface StreamClient {
 
    String INPUT = "myMessageInput";
 
    String OUTPUT = "myMessageOutput";
 
    //接收消息，使用SubscribableChannel 
    @Input(StreamClient.INPUT)
    SubscribableChannel input();
 
    //发送消息，使用MessageChannel 
    @Output(StreamClient.OUTPUT)
    MessageChannel output();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

新建消息接收端 StreamReceiver

package com.viki.order.message;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Component;
 
/**
 * @Author Sakura
 * @Date 10/11/2019
 **/
@Component
@Slf4j
@EnableBinding(StreamClient.class) //上边定义的接口文件
public class StreamReceiver {
 
    @StreamListener(StreamClient.OUTPUT)
    public void processOutput(Object message){
        log.info("StreamReceiver output : {}", message);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

新建发送端controller

package com.viki.order.controller;
 
import com.viki.order.dto.OrderDTO;
import com.viki.order.message.StreamClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.jaxb.SpringDataJaxb;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.util.Date;
 
/**
 * @Author Sakura
 * @Date 10/11/2019
 **/
@RestController
public class SendMessageController {
 
    @Autowired
    private StreamClient streamClient;
 
    @GetMapping("/sendMessage")
    public void sendMessage(){
        String message = "send to mq: now is " + new Date();
        streamClient.output().send(MessageBuilder.withPayload(message).build());
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

此时StreamClient 调用的output方法，与接收端StreamListener的值应一致

启动项目，mq就会新增一个queue