Spring Cloud Stream 实现MQ集成

Spring Cloud Stream是一个构建消息驱动微服务的框架，应用程序通过input通道或者output通道来与Spring Cloud Stream中binder交互，通过配置来binding。而Spring Cloud Stream的binder负责与中间件交互，消息的中间件有(RabbitMQ, Kafka, ActiveMQ)。

本文以RabbitMQ作为中间件作为讲解

Producer

（1）通过IDEA的spring init新建Spring Boot项目

引入依赖

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

配置yml文件

server:
  port: 8081
spring:
  application:
    name: producer
  cloud:
    stream:
      bindings:
        output1:
          destination: Theme-People #可以理解为一个和订阅者的通道
          group: Group-Boy
          content-type: application/json   
          producer:
            partitionKeyExpression: headers['partitionKey']
            partitionCount: 2

（2）新建一个实体类用于发送

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class Person {

    private String id;

    private String name;

    private int age;

    private String password;
}

（3）新建输出通道

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;

@Component
public interface OutputInterface {
    String OUTPUT1 = "output1";
    String OUTPUT2 = "output2";
    //创建一个输出通道 output1
    @Output(OUTPUT1)
    MessageChannel output1();
}

（4）新建一个Service用域处理发送消息

import com.cloud.cloud_streaming.vo.Person;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class StreamService {

    @Autowired
    @Qualifier(OutputInterface.OUTPUT1)
    MessageChannel channel1;

    public void sendMessage(String channel) {
        String message = "Send Message Manually,From Channel:" + channel;
        log.info( "Send Message from channel:" + channel );
        switch (channel) {
            case OutputInterface.OUTPUT1:
                channel1.send( MessageBuilder.withPayload( message ).build() );
                return;
            default:
                log.info( "参数错误: " + channel );
                return;
        }
    }
}

（5）新建一个定时任务类

import com.cloud.cloud_streaming.vo.Person;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;

@Slf4j
@EnableBinding(value = {OutputInterface.class})
@EnableScheduling
public class MessageSender {

    @Autowired
    private OutputInterface outputInterface;

    /**
     * 第一种方法, 没有指定output的MessageChannel, 通过OutputInterface去拿具体的Channel
     * 设置partitionKey主要是为了分区用, 可以根据根据这个partitionKey来分区
采用定时任务去发送
     */
    @Scheduled(initialDelay = 1000, fixedRate = 5000)
    public void sendMessageMethod1() {
        Message message = MessageBuilder.withPayload( "From sendMessageMethod1" )
                .setHeader( "partitionKey", 1 )
                .build();

        outputInterface.output1().send( message );
    }

    /**
     * 第二种方法, 直接指定output的MessageChannel
     */
//    @Autowired
//    @Qualifier(OutputInterface.OUTPUT2)
//    MessageChannel output;
//
//    @Scheduled(initialDelay = 2000, fixedRate = 4000)
//    public void sendMessageMethod2() {
//        Person p = new Person();
//        p.setName( "Person2" );
//        p.setAge( 1 );
//
//        output.send( MessageBuilder.withPayload( p )
//                .setHeader( "partitionKey", 2 )
//                .build() );
//    }
}

（6）Controller层

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@Slf4j
public class StreamController {
    @Autowired
    StreamService streamService;

    @GetMapping("/sendMessageByChannalName")
    public void sendMessage(String channel) {
        streamService.sendMessage( channel );
    }
}

Cusomer

（1）引入依赖

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    <version>2.1.2.RELEASE</version>
</dependency>

（2）配置yml

spring:
  application:
    name: consumer
  cloud:
    stream:
      bindings:
        input1:
          destination: Theme-Animal  #h对应的输出通道 同一个通道
          group: Group-Boy
          content-type: application/json
          consumer:
            partitioned: true #是否分区
      instance-index: 0  #用那个分区来接收消息
      instance-count: 2  #分区数

（3）输入类

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;

@Component
public interface InputInterface {
    String INPUT1 = "input1";

    @Input(INPUT1)
    SubscribableChannel input1();
}

（4）同样新建一个实体类，因为采用的是发送实体

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class Person {
    private String name;

    private int age;
}

（5）接受消息的处理层

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;

@Slf4j
@EnableBinding(value = {InputInterface.class})
public class MessageReceiver {
    @StreamListener(InputInterface.INPUT1)
    public void receiveMessageFromChannel1(@Payload Message <String> payload, @Header(AmqpHeaders.CONSUMER_QUEUE) String partition) {
        log.info( "ReceiveMessageFrom INPUT1, message: {}, Queue:{}", payload.getPayload(), partition );
    }
}

启动

• 启动Producer的project
• 再启动Cusomer的project