序言

之前的開發(fā)主要是底層開發(fā)，沒有深入涉及到消息方面。現(xiàn)在面對的是一個這樣的場景：
假設(shè)公司項(xiàng)目A用了RabbitMQ，而項(xiàng)目B用了Kafka。這時候就會出現(xiàn)有兩個消息框架，這兩個消息框架可能編碼有所不同，且結(jié)構(gòu)也有所不同，而且之前甚至可能使用的是別的框架，造成了一個不易管理的局面。目前我的需求是不改動或者說少量代碼完成兩個消息隊(duì)列之間的切換。我要屏蔽掉切換的成本。

spring cloud stream官方文檔

PS：如有英文，是作者純純的懶，懶得翻譯

消息隊(duì)列

市面上大部分消息隊(duì)列的格局應(yīng)該是 生產(chǎn)者 -》 broker -》消費(fèi)者
采用的是發(fā)布-訂閱的模式，大概的元素有如下幾個：
Message：生產(chǎn)者/消費(fèi)者之間靠消息媒介傳遞信息內(nèi)容
MessageChannel：消息必須走特定的通道
隊(duì)列：假如發(fā)消息會先發(fā)到消息隊(duì)列當(dāng)中

cloud Stream設(shè)計(jì)

通過定義綁定器Binder作為中間層，實(shí)現(xiàn)了應(yīng)用程序與消息中間件細(xì)節(jié)之間的隔離。Binder可以生成Binding，Binding用來綁定消息容器的生產(chǎn)者和消費(fèi)者，它有兩種類型，INPUT和OUTPUT，INPUT對應(yīng)于消費(fèi)者，OUTPUT對應(yīng)于生產(chǎn)者。

用Stream鏈接rabbitMq

我們需要選用一個mq，我們使用一個測試環(huán)境的mq用于學(xué)習(xí)demo，我們需要引入一些依賴。

        <!-- https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-starter-stream-rabbit -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
            <version>3.2.5</version>
        </dependency>

Stream中的消息通信方式遵循了發(fā)布-訂閱模式，Topic主題進(jìn)行廣播，在RabbitMQ就是Exchange，在Kakfa中就是Topic。

核心組件

Binder： 很方便的連接中間件,屏蔽差異
Channel： 通道，是隊(duì)列Queue的一種抽象，在消息通訊系統(tǒng)中就是實(shí)現(xiàn)存儲和轉(zhuǎn)發(fā)的媒介，通過Channe對隊(duì)列進(jìn)行配置
Source（源：發(fā)送者）和Sink（水槽：接受者）： 簡單的可理解為參照對象是Spring Cloud Stream自身，從Stream發(fā)布消息就是輸出，接受消息就是輸入。

注解

@EnableBinding

通過將@EnableBinding注釋應(yīng)用于應(yīng)用程序的一個配置類，可以將Spring應(yīng)用程序轉(zhuǎn)換為Spring Cloud Stream應(yīng)用程序。@EnableBinding注釋本身使用@Configuration進(jìn)行元注釋，并觸發(fā)Spring Cloud Stream基礎(chǔ)設(shè)施的配置：

...
@Import(...)
@Configuration
@EnableIntegration
public @interface EnableBinding {
    ...
    Class<?>[] value() default {};
}

@EnableBinding注釋可以將一個或多個接口類作為參數(shù)，這些接口類包含表示可綁定組件（通常是消息通道）的方法。

使用案例

@EnableBinding({XXXStreamClient.class})

@Input and @Output

Spring Cloud Stream應(yīng)用程序可以在接口中定義任意數(shù)量的輸入和輸出通道，分別為@input和@output方法：

public interface Barista {

    @Input
    SubscribableChannel orders();

    @Output
    MessageChannel hotDrinks();

    @Output
    MessageChannel coldDrinks();
}

使用此接口作為@EnableBinding的參數(shù)將觸發(fā)創(chuàng)建三個綁定通道，分別命名為orders、hotDrinks和coldDrinks。

@EnableBinding(Barista.class)
public class CafeConfiguration {

   ...
}

Customizing Channel Names

使用@Input和@Output注釋，可以為通道指定自定義通道名稱，如以下示例所示：
public interface Barista {
…
@Input(“inboundOrders”)
SubscribableChannel orders();
}

In this example, the created bound channel will be named inboundOrders.

Source, Sink, and Processor

為了簡單地解決最常見的用例（包括輸入通道、輸出通道或兩者），Spring Cloud Stream提供了三個開箱即用的預(yù)定義接口。

Source

Source 可以用于具有單個出站通道的應(yīng)用程序。

public interface Source {

  String OUTPUT = "output";

  @Output(Source.OUTPUT)
  MessageChannel output();

}

Sink

Sink can be used for an application which has a single inbound channel.

public interface Sink {

  String INPUT = "input";

  @Input(Sink.INPUT)
  SubscribableChannel input();

}

Processor

Processor can be used for an application which has both an inbound channel and an outbound channel.

public interface Processor extends Source, Sink {
}

Accessing Bound Channels

每個綁定的接口，Spring Cloud Stream將生成一個實(shí)現(xiàn)該接口的bean。調(diào)用其中一個bean的@Input 或@Output 方法將返回相關(guān)的綁定通道。
以下示例中的bean在調(diào)用其hello方法時在輸出通道上發(fā)送消息。它在注入的源bean上調(diào)用output（）來檢索目標(biāo)通道。

@Component
public class SendingBean {

    private Source source;

    @Autowired
    public SendingBean(Source source) {
        this.source = source;
    }

    public void sayHello(String name) {
         source.output().send(MessageBuilder.withPayload(name).build());
    }
}

Bound channels can be also injected directly:

@Component
public class SendingBean {

    private MessageChannel output;

    @Autowired
    public SendingBean(MessageChannel output) {
        this.output = output;
    }

    public void sayHello(String name) {
         output.send(MessageBuilder.withPayload(name).build());
    }
}

If the name of the channel is customized on the declaring annotation, that name should be used instead of the method name. Given the following declaration:

public interface CustomSource {
    ...
    @Output("customOutput")
    MessageChannel output();
}

The channel will be injected as shown in the following example:

@Component
public class SendingBean {

    private MessageChannel output;

    @Autowired
    public SendingBean(@Qualifier("customOutput") MessageChannel output) {
        this.output = output;
    }

    public void sayHello(String name) {
         this.output.send(MessageBuilder.withPayload(name).build());
    }
}

Using @StreamListener for Automatic Content Type Handling

作為對Spring Integration支持的補(bǔ)充，Spring Cloud Stream提供了自己的@StreamListener注釋，該注釋以其他Spring消息傳遞注釋（例如，@MessageMapping、@JmsListener、@RabbitListener等）為模型。@StreamListener注釋為處理入站消息提供了一個更簡單的模型，尤其是在處理涉及內(nèi)容類型管理和類型強(qiáng)制的用例時。
Spring Cloud Stream提供了一個可擴(kuò)展的MessageConverter機(jī)制，用于處理綁定通道的數(shù)據(jù)轉(zhuǎn)換，在本例中，用于向用@StreamListener注釋的方法進(jìn)行調(diào)度。以下是處理外部投票事件的應(yīng)用程序示例：

@EnableBinding(Sink.class)
public class VoteHandler {

  @Autowired
  VotingService votingService;

  @StreamListener(Sink.INPUT)
  public void handle(Vote vote) {
    votingService.record(vote);
  }
}

當(dāng)考慮具有String有效載荷和application/json的contentType標(biāo)頭的入站消息時，可以看到@StreamListener和Spring Integration@ServiceActivator之間的區(qū)別。在@StreamListener的情況下，MessageConverter機(jī)制將使用contentType標(biāo)頭將String有效載荷解析為Vote對象。
與其他Spring Messaging方法一樣，方法參數(shù)可以用@Payload、@Headers和@Header進(jìn)行注釋。
對于返回數(shù)據(jù)的方法，必須使用@SendTo注釋為該方法返回的數(shù)據(jù)指定輸出綁定目標(biāo)：

@EnableBinding(Processor.class)
public class TransformProcessor {

  @Autowired
  VotingService votingService;

  @StreamListener(Processor.INPUT)
  @SendTo(Processor.OUTPUT)
  public VoteResult handle(Vote vote) {
    return votingService.record(vote);
  }
}

使用@StreamListener將消息調(diào)度到多個方法
在本例中，所有帶有值為foo的頭類型的消息都將被調(diào)度到receiveFoo方法，而所有帶有值bar的頭類型消息都將調(diào)度到receive bar方法。

@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class TestPojoWithAnnotatedArguments {

    @StreamListener(target = Sink.INPUT, condition = "headers['type']=='foo'")
    public void receiveFoo(@Payload FooPojo fooPojo) {
       // handle the message
    }

    @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bar'")
    public void receiveBar(@Payload BarPojo barPojo) {
       // handle the message
    }
}

demo

先安裝rabbitMq

過程 略

代碼部分

配置

server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: # 在此處配置要綁定的rabbitmq的服務(wù)信息；
        defaultRabbit: # 表示定義的名稱，用于于binding整合
          type: rabbit # 消息組件類型
          environment: # 設(shè)置rabbitmq的相關(guān)的環(huán)境配置
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
          test: # 表示定義的名稱，用于于binding整合
            type: rabbit # 消息組件類型
            environment: # 設(shè)置rabbitmq的相關(guān)的環(huán)境配置
              spring:
                rabbitmq:
                  host: localhost
                  port: 5672
                  username: guest
                  password: guest
      bindings: # 服務(wù)的整合處理
        testOutput: #生產(chǎn)者消息輸出通道 ---> 消息輸出通道 = 生產(chǎn)者相關(guān)的定義：Exchange & Queue
          destination: exchange-test          #exchange名稱,交換模式默認(rèn)是topic;把SpringCloud Stream的消息輸出通道綁定到RabbitMQ的exchange-test交換器。
          content-type: application/json      #設(shè)置消息的類型，本次為json
          default-binder: defaultRabbit       #設(shè)置要綁定的消息服務(wù)的具體設(shè)置，默認(rèn)綁定RabbitMQ
          group: testGroup                    #分組=Queue名稱，如果不設(shè)置會使用默認(rèn)的組流水號
        testInput: #消費(fèi)者消息輸入通道 ---> 消息輸入通道 = 消費(fèi)者相關(guān)的定義：Exchange & Queue
            destination: exchange-test          #exchange名稱,交換模式默認(rèn)是topic;把SpringCloud Stream的消息輸入通道綁定到RabbitMQ的exchange-test交換器。
            content-type: application/json
            default-binder: defaultRabbit
            group: testGroup
        testOutput1: #生產(chǎn)者消息輸出通道 ---> 消息輸出通道 = 生產(chǎn)者相關(guān)的定義：Exchange & Queue
          destination: exchange-test          #exchange名稱,交換模式默認(rèn)是topic;把SpringCloud Stream的消息輸出通道綁定到RabbitMQ的exchange-test交換器。
          content-type: application/json      #設(shè)置消息的類型，本次為json
          default-binder: test       #設(shè)置要綁定的消息服務(wù)的具體設(shè)置，默認(rèn)綁定RabbitMQ
          group: testGroup1                    #分組=Queue名稱，如果不設(shè)置會使用默認(rèn)的組流水號
        testInput1: #消費(fèi)者消息輸入通道 ---> 消息輸入通道 = 消費(fèi)者相關(guān)的定義：Exchange & Queue
          destination: exchange-test          #exchange名稱,交換模式默認(rèn)是topic;把SpringCloud Stream的消息輸入通道綁定到RabbitMQ的exchange-test交換器。
          content-type: application/json
          default-binder: test
          group: testGroup1

TestChannelProcessor

package com.anxin.rabbitmq_scz.provider;

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

@Component
public interface TestChannelProcessor {
    /**
     * 生產(chǎn)者消息輸出通道（需要與配置文件中的保持一致）
     */
    String TEST_OUTPUT = "testOutput";

    /**
     * 消息生產(chǎn)
     *
     * @return
     */
    @Output(TEST_OUTPUT)
    MessageChannel testOutput();

    /**
     * 消費(fèi)者消息輸入通道（需要與配置文件中的保持一致）
     */
    String TEST_INPUT = "testInput";

    /**
     * 消息消費(fèi)
     *
     * @return
     */
    @Input(TEST_INPUT)
    SubscribableChannel testInput();
}

TestMessageProducer

package com.anxin.rabbitmq_scz.provider;

import com.alibaba.fastjson.JSON;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;

import java.util.HashMap;
import java.util.Map;

@EnableBinding(value = {TestChannelProcessor.class})
public class TestMessageProducer {
    @Autowired
    private BinderAwareChannelResolver channelResolver;

    /**
     * 生產(chǎn)消息
     *
     * @param msg
     */
    public void testSendMessage(String msg) {
        Map<String, Object> headers = new HashMap<>();
        Map<String, Object> payload = new HashMap<>();
        payload.put("msg", msg);
        System.err.println("生產(chǎn)者發(fā)送消息：" + JSON.toJSONString(payload));
        channelResolver.resolveDestination(TestChannelProcessor.TEST_OUTPUT).send(
                MessageBuilder.createMessage(payload, new MessageHeaders(headers))
        );
    }

    /**
     * 生產(chǎn)消息
     *
     * @param msg
     */
    public void testSendMessage1(String msg) {
        Map<String, Object> headers = new HashMap<>();
        Map<String, Object> payload = new HashMap<>();
        payload.put("msg", msg);
        System.err.println("生產(chǎn)者發(fā)送消息：" + JSON.toJSONString(payload));
        channelResolver.resolveDestination(TestChannelProcessor1.TEST_OUTPUT).send(
                MessageBuilder.createMessage(payload, new MessageHeaders(headers))
        );
    }
}

TestMessageConsumer

package com.anxin.rabbitmq_scz.consumer;

import com.anxin.rabbitmq_scz.provider.TestChannelProcessor;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

@EnableBinding(TestChannelProcessor.class)
public class TestMessageConsumer {
    @StreamListener(TestChannelProcessor.TEST_INPUT)
    public void testConsumeMessage(Message<String> message) {
        System.err.println("消費(fèi)者消費(fèi)消息：" + message.getPayload());
    }
}

SwaggerConfig

package com.anxin.rabbitmq_scz.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.context.request.async.DeferredResult;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.builders.PathSelectors;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;

@Configuration
@EnableSwagger2
public class SwaggerConfig {

    @Bean
    public Docket createRestApi() {

        return new Docket(DocumentationType.SWAGGER_2)
                .genericModelSubstitutes(DeferredResult.class)
                .select()
                .paths(PathSelectors.any())
                .build().apiInfo(apiInfo());

    }

    private ApiInfo apiInfo() {

        return new ApiInfoBuilder().title("Stream server")
                .description("測試SpringCloudStream")
                .termsOfServiceUrl("https://spring.io/projects/spring-cloud-stream")
                .version("1.0").build();
    }
}

TestController

package com.anxin.rabbitmq_scz.controller;

import com.anxin.rabbitmq_scz.provider.TestMessageProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class TestController {

    @Autowired
    private TestMessageProducer testMessageProducer;

    /**
     * 發(fā)送保存訂單消息
     *
     * @param message
     */
    @GetMapping(value = "sendTestMessage")
    public void sendTestMessage(@RequestParam("message") String message) {
        //發(fā)送消息
        testMessageProducer.testSendMessage(message);
    }

    /**
     * 發(fā)送保存訂單消息
     *
     * @param message
     */
    @GetMapping(value = "sendTestMessage1")
    public void sendTestMessage1(@RequestParam("message") String message) {
        //發(fā)送消息
        testMessageProducer.testSendMessage1(message);
    }
}

啟動即可

結(jié)束語

新入一門技術(shù)最簡單最快的辦法就是觀看它的官方文檔，學(xué)會api的調(diào)用，代價是極低的，但是如果要深入一門技術(shù)，必須需要閱讀其源碼且結(jié)合其設(shè)計(jì)模式。文章來源地址http://www.zghlxwxcb.cn/news/detail-755929.html

到了這里，關(guān)于Spring cloud stream 結(jié)合 rabbitMq使用的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！