RocketMQ學習分享

消息隊列的流派

 

什么是 MQ

Message Queue（MQ），消息隊列中間件。很多人都說：MQ 通過將消息的發送和接收分離來實現應用程序的異步和解偶，這個給人的直覺是——MQ 是異步的，用來解耦的，但是這個只是 MQ 的效果而不是目的。MQ 真正的目的是為了通訊，屏蔽底層復雜的通訊協議，定義了一套應用層的、更加簡單的通訊協議。一個分布式系統中兩個模塊之間通訊要么是 HTTP，要么是自己開發的 TCP，但是這兩種協議其實都是原始的協議。HTTP 協議很難實現兩端通訊——模塊 A 可以調用 B，B 也可以主動調用 A，如果要做到這個兩端都要背上 WebServer，而且還不支持長連接（HTTP 2.0 的庫根本找不到）。TCP 就更加原始了，粘包、心跳、私有的協議，想一想頭皮就發麻。MQ 所要做的就是在這些協議之上構建一個簡單的“協議”——生產者/消費者模型。MQ 帶給我的“協議”不是具體的通訊協議，而是更高層次通訊模型。它定義了兩個對象——發送數據的叫生產者；接收數據的叫消費者， 提供一個 SDK 讓我們可以定義自己的生產者和消費者實現消息通訊而無視底層通訊協議

 

有 Broker 的 MQ

這個流派通常有一台服務器作為 Broker，所有的消息都通過它中轉。生產者把消息發送給它就結束自己的任務了，Broker 則把消息主動推送給消費者（或者消費者主動輪詢）

重 Topic

kafka、JMS（ActiveMQ）就屬於這個流派，生產者會發送 key 和數據到 Broker，由 Broker 比較 key 之后決定給哪個消費者。這種模式是我們最常見的模式，是我們對 MQ 最多的印象。在這種模式下一個 topic 往往是一個比較大的概念，甚至一個系統中就可能只有一個 topic，topic 某種意義上就是 queue，生產者發送 key 相當於說：“hi，把數據放到 key 的隊列中”

如上圖所示，Broker 定義了三個隊列，key1，key2，key3，生產者發送數據的時候會發送 key1 和 data，Broker 在推送數據的時候則推送 data（也可能把 key 帶上）。

雖然架構一樣但是 kafka 的性能要比 jms 的性能不知道高到多少倍，所以基本這種類型的 MQ 只有 kafka 一種備選方案。如果你需要一條暴力的數據流（在乎性能而非靈活性）那么 kafka 是最好的選擇

輕 Topic

這種的代表是 RabbitMQ（或者說是 AMQP）。生產者發送 key 和數據，消費者定義訂閱的隊列，Broker 收到數據之后會通過一定的邏輯計算出 key 對應的隊列，然后把數據交給隊列

這種模式下解耦了 key 和 queue，在這種架構中 queue 是非常輕量級的（在 RabbitMQ 中它的上限取決於你的內存），消費者關心的只是自己的 queue；生產者不必關心數據最終給誰只要指定 key 就行了，中間的那層映射在 AMQP 中叫 exchange（交換機）。

AMQP 中有四種 exchange

• Direct exchange：key 就等於 queue
• Fanout exchange：無視 key，給所有的 queue 都來一份
• Topic exchange：key 可以用“寬字符”模糊匹配 queue
• Headers exchange：無視 key，通過查看消息的頭部元數據來決定發給那個 queue（AMQP 頭部元數據非常豐富而且可以自定義）

這種結構的架構給通訊帶來了很大的靈活性，我們能想到的通訊方式都可以用這四種 exchange 表達出來。如果你需要一個企業數據總線（在乎靈活性）那么 RabbitMQ 絕對的值得一用

 

無 Broker 的 MQ

無 Broker 的 MQ 的代表是 ZeroMQ。該作者非常睿智，他非常敏銳的意識到——MQ 是更高級的 Socket，它是解決通訊問題的。所以 ZeroMQ 被設計成了一個“庫”而不是一個中間件，這種實現也可以達到——沒有 Broker 的目的：

節點之間通訊的消息都是發送到彼此的隊列中，每個節點都既是生產者又是消費者。ZeroMQ 做的事情就是封裝出一套類似於 Socket 的 API 可以完成發送數據，讀取數據

ZeroMQ 其實就是一個跨語言的、重量級的 Actor 模型郵箱庫。你可以把自己的程序想象成一個 Actor，ZeroMQ 就是提供郵箱功能的庫；ZeroMQ 可以實現同一台機器的 RPC 通訊也可以實現不同機器的 TCP、UDP 通訊，如果你需要一個強大的、靈活、野蠻的通訊能力，別猶豫 ZeroMQ

 

附：Queue 和 Topic 的區別

• Queue： 一個發布者發布消息，下面的接收者按隊列順序接收，比如發布了 10 個消息，兩個接收者 A,B 那就是 A,B 總共 會收到 10 條消息，不重復。
• Topic： 一個發布者發布消息，有兩個接收者 A,B 來訂閱，那么發布了 10 條消息，A,B 各收到 10 條消息。

 

 

RocketMQ 簡介

 

概述

消息隊列作為高並發系統的核心組件之一，能夠幫助業務系統解構提升開發效率和系統穩定性。主要具有以下優勢：

• 削峰填谷： 主要解決瞬時寫壓力大於應用服務能力導致消息丟失、系統奔潰等問題
• 系統解耦： 解決不同重要程度、不同能力級別系統之間依賴導致一死全死
• 提升性能： 當存在一對多調用時，可以發一條消息給消息系統，讓消息系統通知相關系統
• 蓄流壓測： 線上有些鏈路不好壓測，可以通過堆積一定量消息再放開來壓測

 

RocketMQ

Apache Alibaba RocketMQ 是一個消息中間件。消息中間件中有兩個角色：消息生產者和消息消費者。RocketMQ 里同樣有這兩個概念，消息生產者負責創建消息並發送到 RocketMQ 服務器，RocketMQ 服務器會將消息持久化到磁盤，消息消費者從 RocketMQ 服務器拉取消息並提交給應用消費。

 

RocketMQ 特點

RocketMQ 是一款分布式、隊列模型的消息中間件，具有以下特點：

• 支持嚴格的消息順序
• 支持 Topic 與 Queue 兩種模式
• 億級消息堆積能力
• 比較友好的分布式特性
• 同時支持 Push 與 Pull 方式消費消息
• 歷經多次天貓雙十一海量消息考驗

 

RocketMQ 優勢

目前主流的 MQ 主要是 RocketMQ、kafka、RabbitMQ，其主要優勢有：

• 支持事務型消息（消息發送和 DB 操作保持兩方的最終一致性，RabbitMQ 和 Kafka 不支持）
• 支持結合 RocketMQ 的多個系統之間數據最終一致性（多方事務，二方事務是前提）
• 支持 18 個級別的延遲消息（RabbitMQ 和 Kafka 不支持）
• 支持指定次數和時間間隔的失敗消息重發（Kafka 不支持，RabbitMQ 需要手動確認）
• 支持 Consumer 端 Tag 過濾，減少不必要的網絡傳輸（RabbitMQ 和 Kafka 不支持）
• 支持重復消費（RabbitMQ 不支持，Kafka 支持）

 

消息隊列對比參照表

 

基於 Docker 安裝 RocketMQ

 

docker-compose.yml

注意：啟動 RocketMQ Server + Broker + Console 至少需要 2G 內存

version: '3.5'
services:
  rmqnamesrv:
    image: foxiswho/rocketmq:server
    container_name: rmqnamesrv
    ports:
      - 9876:9876
    volumes:
      - ./data/logs:/opt/logs
      - ./data/store:/opt/store
    networks:
        rmq:
          aliases:
            - rmqnamesrv

  rmqbroker:
    image: foxiswho/rocketmq:broker
    container_name: rmqbroker
    ports:
      - 10909:10909
      - 10911:10911
    volumes:
      - ./data/logs:/opt/logs
      - ./data/store:/opt/store
      - ./data/brokerconf/broker.conf:/etc/rocketmq/broker.conf
    environment:
        NAMESRV_ADDR: "rmqnamesrv:9876"
        JAVA_OPTS: " -Duser.home=/opt"
        JAVA_OPT_EXT: "-server -Xms128m -Xmx128m -Xmn128m"
    command: mqbroker -c /etc/rocketmq/broker.conf
    depends_on:
      - rmqnamesrv
    networks:
      rmq:
        aliases:
          - rmqbroker

  rmqconsole:
    image: styletang/rocketmq-console-ng
    container_name: rmqconsole
    ports:
      - 8080:8080
    environment:
        JAVA_OPTS: "-Drocketmq.namesrv.addr=rmqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false"
    depends_on:
      - rmqnamesrv
    networks:
      rmq:
        aliases:
          - rmqconsole

networks:
  rmq:
    name: rmq
    driver: bridge

 

broker.conf

RocketMQ Broker 需要一個配置文件，按照上面的 Compose 配置，我們需要在 ./data/brokerconf/ 目錄下創建一個名為 broker.conf 的配置文件，內容如下：

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.


# 所屬集群名字
brokerClusterName=DefaultCluster

# broker 名字，注意此處不同的配置文件填寫的不一樣，如果在 broker-a.properties 使用: broker-a,
# 在 broker-b.properties 使用: broker-b
brokerName=broker-a

# 0 表示 Master，> 0 表示 Slave
brokerId=0

# nameServer地址，分號分割
# namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876

# 啟動IP,如果 docker 報 com.alibaba.rocketmq.remoting.exception.RemotingConnectException: connect to <192.168.0.120:10909> failed
# 解決方式1 加上一句 producer.setVipChannelEnabled(false);，解決方式2 brokerIP1 設置宿主機IP，不要使用docker 內部IP
# brokerIP1=192.168.0.253

# 在發送消息時，自動創建服務器不存在的topic，默認創建的隊列數
defaultTopicQueueNums=4

# 是否允許 Broker 自動創建 Topic，建議線下開啟，線上關閉 ！！！這里仔細看是 false，false，false
autoCreateTopicEnable=true

# 是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉
autoCreateSubscriptionGroup=true

# Broker 對外服務的監聽端口
listenPort=10911

# 刪除文件時間點，默認凌晨4點
deleteWhen=04

# 文件保留時間，默認48小時
fileReservedTime=120

# commitLog 每個文件的大小默認1G
mapedFileSizeCommitLog=1073741824

# ConsumeQueue 每個文件默認存 30W 條，根據業務情況調整
mapedFileSizeConsumeQueue=300000

# destroyMapedFileIntervalForcibly=120000
# redeleteHangedFileInterval=120000
# 檢測物理文件磁盤空間
diskMaxUsedSpaceRatio=88
# 存儲路徑
# storePathRootDir=/home/ztztdata/rocketmq-all-4.1.0-incubating/store
# commitLog 存儲路徑
# storePathCommitLog=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/commitlog
# 消費隊列存儲
# storePathConsumeQueue=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/consumequeue
# 消息索引存儲路徑
# storePathIndex=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/index
# checkpoint 文件存儲路徑
# storeCheckpoint=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/checkpoint
# abort 文件存儲路徑
# abortFile=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/abort
# 限制的消息大小
maxMessageSize=65536

# flushCommitLogLeastPages=4
# flushConsumeQueueLeastPages=2
# flushCommitLogThoroughInterval=10000
# flushConsumeQueueThoroughInterval=60000

# Broker 的角色
# - ASYNC_MASTER 異步復制Master
# - SYNC_MASTER 同步雙寫Master
# - SLAVE
brokerRole=ASYNC_MASTER

# 刷盤方式
# - ASYNC_FLUSH 異步刷盤
# - SYNC_FLUSH 同步刷盤
flushDiskType=ASYNC_FLUSH

# 發消息線程池數量
# sendMessageThreadPoolNums=128
# 拉消息線程池數量
# pullMessageThreadPoolNums=128

 

RocketMQ 控制台

訪問 http://rmqIP:8080 登入控制台

 

RocketMQ 生產者

 

概述

RocketMQ 是一款開源的分布式消息系統，基於高可用分布式集群技術，提供低延時的、高可靠的消息發布與訂閱服務。

由於本教程整個案例基於 Spring Cloud，故我們采用 Spring Cloud Stream 完成一次發布和訂閱

官方教程

 

Spring Cloud Stream

Spring Cloud Stream 是一個用於構建基於消息的微服務應用框架。它基於 Spring Boot 來創建具有生產級別的單機 Spring 應用，並且使用 Spring Integration 與 Broker 進行連接。

Spring Cloud Stream 提供了消息中間件配置的統一抽象，推出了 publish-subscribe、consumer groups、partition 這些統一的概念。

Spring Cloud Stream 內部有兩個概念：

• Binder： 跟外部消息中間件集成的組件，用來創建 Binding，各消息中間件都有自己的 Binder 實現。
• Binding： 包括 Input Binding 和 Output Binding。

Binding 在消息中間件與應用程序提供的 Provider 和 Consumer 之間提供了一個橋梁，實現了開發者只需使用應用程序的 Provider 或 Consumer 生產或消費數據即可，屏蔽了開發者與底層消息中間件的接觸。

 

解決連接超時問題

在之前的章節中，我們采用 Docker 部署了 RocketMQ 服務，此時 RocketMQ Broker 暴露的地址和端口(10909，10911)是基於容器的，會導致我們開發機無法連接，從而引發 org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout 異常

注意下圖中的 IP 地址，這個是容器的 IP，開發機與容器不在一個局域網所以無法連接。

解決方案是在 broker.conf 配置文件中增加 brokerIP1=宿主機IP 即可

 

POM

主要增加了 org.springframework.cloud:spring-cloud-starter-stream-rocketmq 依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>com.funtl</groupId>
        <artifactId>hello-spring-cloud-alibaba-dependencies</artifactId>
        <version>1.0.0-SNAPSHOT</version>
        <relativePath>../hello-spring-cloud-alibaba-dependencies/pom.xml</relativePath>
    </parent>

    <artifactId>hello-spring-cloud-alibaba-rocketmq-provider</artifactId>
    <packaging>jar</packaging>

    <name>hello-spring-cloud-alibaba-rocketmq-provider</name>
    <url>http://www.funtl.com</url>
    <inceptionYear>2018-Now</inceptionYear>

    <dependencies>
        <!-- Spring Boot Begin -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- Spring Boot End -->

        <!-- Spring Cloud Begin -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
        </dependency>
        <!-- Spring Cloud End -->
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <mainClass>com.funtl.hello.spring.cloud.alibaba.rocketmq.provider.RocketMQProviderApplication</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

 

消息生產者服務

package com.funtl.hello.spring.cloud.alibaba.rocketmq.provider.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class ProviderService {
    @Autowired
    private MessageChannel output;

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

 

Application

配置 Output(Source.class) 的 Binding 信息並配合 @EnableBinding 注解使其生效

package com.funtl.hello.spring.cloud.alibaba.rocketmq.provider;

import com.funtl.hello.spring.cloud.alibaba.rocketmq.provider.service.ProviderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;

@SpringBootApplication
@EnableBinding({Source.class})
public class RocketMQProviderApplication implements CommandLineRunner {

    @Autowired
    private ProviderService providerService;

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

    /**
     * 實現了 CommandLineRunner 接口，只是為了 Spring Boot 啟動時執行任務，不必特別在意
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        providerService.send("Hello RocketMQ");
    }
}

application.yml

spring:
  application:
    name: rocketmq-provider
  cloud:
    stream:
      rocketmq:
        binder:
          # RocketMQ 服務器地址
          namesrv-addr: 192.168.10.149:9876
      bindings:
        # 這里是個 Map 類型參數，{} 為 YAML 中 Map 的行內寫法
        output: {destination: test-topic, content-type: application/json}

server:
  port: 9093

management:
  endpoints:
    web:
      exposure:
        include: '*'

運行成功后即可在 RocketMQ 控制台的 消息 列表中選擇 test-topic 主題即可看到發送的消息。

 

RocketMQ 消費者

 

POM

主要增加了 org.springframework.cloud:spring-cloud-starter-stream-rocketmq 依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>com.funtl</groupId>
        <artifactId>hello-spring-cloud-alibaba-dependencies</artifactId>
        <version>1.0.0-SNAPSHOT</version>
        <relativePath>../hello-spring-cloud-alibaba-dependencies/pom.xml</relativePath>
    </parent>

    <artifactId>hello-spring-cloud-alibaba-rocketmq-consumer</artifactId>
    <packaging>jar</packaging>

    <name>hello-spring-cloud-alibaba-rocketmq-consumer</name>
    <url>http://www.funtl.com</url>
    <inceptionYear>2018-Now</inceptionYear>

    <dependencies>
        <!-- Spring Boot Begin -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- Spring Boot End -->

        <!-- Spring Cloud Begin -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
        </dependency>
        <!-- Spring Cloud End -->
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <mainClass>com.funtl.hello.spring.cloud.alibaba.rocketmq.consumer.RocketMQConsumerApplication</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

消息消費者服務

主要使用 @StreamListener("input") 注解來訂閱從名為 input 的 Binding 中接收的消息

package com.funtl.hello.spring.cloud.alibaba.rocketmq.consumer.receive;

import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;

@Service
public class ConsumerReceive {

    @StreamListener("input")
    public void receiveInput(String message) {
        System.out.println("Receive input: " + message);
    }
}

Application

配置 Input(Sink.class) 的 Binding 信息並配合 @EnableBinding 注解使其生效

package com.funtl.hello.spring.cloud.alibaba.rocketmq.consumer;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;

@SpringBootApplication
@EnableBinding({Sink.class})
public class RocketMQConsumerApplication {
    public static void main(String[] args) {
        SpringApplication.run(RocketMQConsumerApplication.class, args);
    }
}

application.yml

spring:
  application:
    name: rocketmq-consumer
  cloud:
    stream:
      rocketmq:
        binder:
          namesrv-addr: 192.168.10.149:9876
        bindings:
          input: {consumer.orderly: true}
      bindings:
        input: {destination: test-topic, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}

server:
  port: 9094

management:
  endpoints:
    web:
      exposure:
        include: '*'

運行成功后即可在控制台接收到消息：Receive input: Hello RocketMQ

 

RocketMQ 自定義 Binding

 

概述

在實際生產中，我們需要發布和訂閱的消息可能不止一種 Topic ，故此時就需要使用自定義 Binding 來幫我們實現多 Topic 的發布和訂閱功能

 

生產者

自定義 Output 接口，代碼如下：

public interface MySource {
    @Output("output1")
    MessageChannel output1();

    @Output("output2")
    MessageChannel output2();
}

發布消息的案例代碼如下：

@Autowired
private MySource source;

public void send(String msg) throws Exception {
    source.output1().send(MessageBuilder.withPayload(msg).build());
}

消費者

自定義 Input 接口，代碼如下：

public interface MySink {
    @Input("input1")
    SubscribableChannel input1();

    @Input("input2")
    SubscribableChannel input2();

    @Input("input3")
    SubscribableChannel input3();

    @Input("input4")
    SubscribableChannel input4();
}

接收消息的案例代碼如下：

@StreamListener("input1")
public void receiveInput1(String receiveMsg) {
    System.out.println("input1 receive: " + receiveMsg);
}

Application

配置 Input 和 Output 的 Binding 信息並配合 @EnableBinding 注解使其生效，代碼如下：

@SpringBootApplication
@EnableBinding({ MySource.class, MySink.class })
public class RocketMQApplication {
	public static void main(String[] args) {
		SpringApplication.run(RocketMQApplication.class, args);
	}
}

application.yml

生產者

spring:
  application:
    name: rocketmq-provider
  cloud:
    stream:
      rocketmq:
        binder:
          namesrv-addr: 192.168.10.149:9876
      bindings:
        output1: {destination: test-topic1, content-type: application/json}
        output2: {destination: test-topic2, content-type: application/json}

消費者

spring:
  application:
    name: rocketmq-consumer
  cloud:
    stream:
      rocketmq:
        binder:
          namesrv-addr: 192.168.10.149:9876
        bindings:
          input: {consumer.orderly: true}
      bindings:
        input1: {destination: test-topic1, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}
        input2: {destination: test-topic1, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}
        input3: {destination: test-topic2, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}
        input4: {destination: test-topic2, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}

 

#感謝閱覽#

引用：千峰