RocketMQ学习分享
消息队列的流派
什么是 MQ
Message Queue（MQ），消息队列中间件。不少人都说：MQ 经过将消息的发送和接收分离来实现应用程序的异步和解偶，这个给人的直觉是——MQ 是异步的，用来解耦的，可是这个只是 MQ 的效果而不是目的。MQ 真正的目的是为了通信，屏蔽底层复杂的通信协议，定义了一套应用层的、更加简单的通信协议。一个分布式系统中两个模块之间通信要么是 HTTP，要么是本身开发的 TCP，可是这两种协议其实都是原始的协议。HTTP 协议很难实现两端通信——模块 A 能够调用 B，B 也能够主动调用 A，若是要作到这个两端都要背上 WebServer，并且还不支持长链接（HTTP 2.0 的库根本找不到）。TCP 就更加原始了，粘包、心跳、私有的协议，想想头皮就发麻。MQ 所要作的就是在这些协议之上构建一个简单的“协议”——生产者/消费者模型。MQ 带给个人“协议”不是具体的通信协议，而是更高层次通信模型。它定义了两个对象——发送数据的叫生产者；接收数据的叫消费者， 提供一个 SDK 让咱们能够定义本身的生产者和消费者实现消息通信而无视底层通信协议git
有 Broker 的 MQ
这个流派一般有一台服务器做为 Broker，全部的消息都经过它中转。生产者把消息发送给它就结束本身的任务了，Broker 则把消息主动推送给消费者（或者消费者主动轮询）github
重 Topic
kafka、JMS（ActiveMQ）就属于这个流派，生产者会发送 key 和数据到 Broker，由 Broker 比较 key 以后决定给哪一个消费者。这种模式是咱们最多见的模式，是咱们对 MQ 最多的印象。在这种模式下一个 topic 每每是一个比较大的概念，甚至一个系统中就可能只有一个 topic，topic 某种意义上就是 queue，生产者发送 key 至关于说：“hi，把数据放到 key 的队列中”web
如上图所示，Broker 定义了三个队列，key1，key2，key3，生产者发送数据的时候会发送 key1 和 data，Broker 在推送数据的时候则推送 data（也可能把 key 带上）。spring
虽然架构同样可是 kafka 的性能要比 jms 的性能不知道高到多少倍，因此基本这种类型的 MQ 只有 kafka 一种备选方案。若是你须要一条暴力的数据流（在意性能而非灵活性）那么 kafka 是最好的选择docker
轻 Topic
这种的表明是 RabbitMQ（或者说是 AMQP）。生产者发送 key 和数据，消费者定义订阅的队列，Broker 收到数据以后会经过必定的逻辑计算出 key 对应的队列，而后把数据交给队列express
这种模式下解耦了 key 和 queue，在这种架构中 queue 是很是轻量级的（在 RabbitMQ 中它的上限取决于你的内存），消费者关心的只是本身的 queue；生产者没必要关心数据最终给谁只要指定 key 就好了，中间的那层映射在 AMQP 中叫 exchange（交换机）。apache
AMQP 中有四种 exchangejson
- 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 内存
1 version: '3.5' 2 services: 3 rmqnamesrv: 4 image: foxiswho/rocketmq:server 5 container_name: rmqnamesrv 6 ports: 7 - 9876:9876 8 volumes: 9 - ./data/logs:/opt/logs 10 - ./data/store:/opt/store 11 networks: 12 rmq: 13 aliases: 14 - rmqnamesrv 15 16 rmqbroker: 17 image: foxiswho/rocketmq:broker 18 container_name: rmqbroker 19 ports: 20 - 10909:10909 21 - 10911:10911 22 volumes: 23 - ./data/logs:/opt/logs 24 - ./data/store:/opt/store 25 - ./data/brokerconf/broker.conf:/etc/rocketmq/broker.conf 26 environment: 27 NAMESRV_ADDR: "rmqnamesrv:9876" 28 JAVA_OPTS: " -Duser.home=/opt" 29 JAVA_OPT_EXT: "-server -Xms128m -Xmx128m -Xmn128m" 30 command: mqbroker -c /etc/rocketmq/broker.conf 31 depends_on: 32 - rmqnamesrv 33 networks: 34 rmq: 35 aliases: 36 - rmqbroker 37 38 rmqconsole: 39 image: styletang/rocketmq-console-ng 40 container_name: rmqconsole 41 ports: 42 - 8080:8080 43 environment: 44 JAVA_OPTS: "-Drocketmq.namesrv.addr=rmqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" 45 depends_on: 46 - rmqnamesrv 47 networks: 48 rmq: 49 aliases: 50 - rmqconsole 51 52 networks: 53 rmq: 54 name: rmq 55 driver: bridge
broker.conf
RocketMQ Broker 须要一个配置文件，按照上面的 Compose 配置，咱们须要在 ./data/brokerconf/ 目录下建立一个名为 broker.conf 的配置文件，内容以下：
1 # Licensed to the Apache Software Foundation (ASF) under one or more 2 # contributor license agreements. See the NOTICE file distributed with 3 # this work for additional information regarding copyright ownership. 4 # The ASF licenses this file to You under the Apache License, Version 2.0 5 # (the "License"); you may not use this file except in compliance with 6 # the License. You may obtain a copy of the License at 7 # 8 # http://www.apache.org/licenses/LICENSE-2.0 9 # 10 # Unless required by applicable law or agreed to in writing, software 11 # distributed under the License is distributed on an "AS IS" BASIS, 12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 # See the License for the specific language governing permissions and 14 # limitations under the License. 15 16 17 # 所属集群名字 18 brokerClusterName=DefaultCluster 19 20 # broker 名字，注意此处不一样的配置文件填写的不同，若是在 broker-a.properties 使用: broker-a, 21 # 在 broker-b.properties 使用: broker-b 22 brokerName=broker-a 23 24 # 0 表示 Master，> 0 表示 Slave 25 brokerId=0 26 27 # nameServer地址，分号分割 28 # namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876 29 30 # 启动IP,若是 docker 报 com.alibaba.rocketmq.remoting.exception.RemotingConnectException: connect to <192.168.0.120:10909> failed 31 # 解决方式1 加上一句 producer.setVipChannelEnabled(false);，解决方式2 brokerIP1 设置宿主机IP，不要使用docker 内部IP 32 # brokerIP1=192.168.0.253 33 34 # 在发送消息时，自动建立服务器不存在的topic，默认建立的队列数 35 defaultTopicQueueNums=4 36 37 # 是否容许 Broker 自动建立 Topic，建议线下开启，线上关闭 ！！！这里仔细看是 false，false，false 38 autoCreateTopicEnable=true 39 40 # 是否容许 Broker 自动建立订阅组，建议线下开启，线上关闭 41 autoCreateSubscriptionGroup=true 42 43 # Broker 对外服务的监听端口 44 listenPort=10911 45 46 # 删除文件时间点，默认凌晨4点 47 deleteWhen=04 48 49 # 文件保留时间，默认48小时 50 fileReservedTime=120 51 52 # commitLog 每一个文件的大小默认1G 53 mapedFileSizeCommitLog=1073741824 54 55 # ConsumeQueue 每一个文件默认存 30W 条，根据业务状况调整 56 mapedFileSizeConsumeQueue=300000 57 58 # destroyMapedFileIntervalForcibly=120000 59 # redeleteHangedFileInterval=120000 60 # 检测物理文件磁盘空间 61 diskMaxUsedSpaceRatio=88 62 # 存储路径 63 # storePathRootDir=/home/ztztdata/rocketmq-all-4.1.0-incubating/store 64 # commitLog 存储路径 65 # storePathCommitLog=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/commitlog 66 # 消费队列存储 67 # storePathConsumeQueue=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/consumequeue 68 # 消息索引存储路径 69 # storePathIndex=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/index 70 # checkpoint 文件存储路径 71 # storeCheckpoint=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/checkpoint 72 # abort 文件存储路径 73 # abortFile=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/abort 74 # 限制的消息大小 75 maxMessageSize=65536 76 77 # flushCommitLogLeastPages=4 78 # flushConsumeQueueLeastPages=2 79 # flushCommitLogThoroughInterval=10000 80 # flushConsumeQueueThoroughInterval=60000 81 82 # Broker 的角色 83 # - ASYNC_MASTER 异步复制Master 84 # - SYNC_MASTER 同步双写Master 85 # - SLAVE 86 brokerRole=ASYNC_MASTER 87 88 # 刷盘方式 89 # - ASYNC_FLUSH 异步刷盘 90 # - SYNC_FLUSH 同步刷盘 91 flushDiskType=ASYNC_FLUSH 92 93 # 发消息线程池数量 94 # sendMessageThreadPoolNums=128 95 # 拉消息线程池数量 96 # 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 依赖
1 <?xml version="1.0" encoding="UTF-8"?> 2 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 3 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4 <modelVersion>4.0.0</modelVersion> 5 6 <parent> 7 <groupId>com.funtl</groupId> 8 <artifactId>hello-spring-cloud-alibaba-dependencies</artifactId> 9 <version>1.0.0-SNAPSHOT</version> 10 <relativePath>../hello-spring-cloud-alibaba-dependencies/pom.xml</relativePath> 11 </parent> 12 13 <artifactId>hello-spring-cloud-alibaba-rocketmq-provider</artifactId> 14 <packaging>jar</packaging> 15 16 <name>hello-spring-cloud-alibaba-rocketmq-provider</name> 17 <url>http://www.funtl.com</url> 18 <inceptionYear>2018-Now</inceptionYear> 19 20 <dependencies> 21 <!-- Spring Boot Begin --> 22 <dependency> 23 <groupId>org.springframework.boot</groupId> 24 <artifactId>spring-boot-starter-web</artifactId> 25 </dependency> 26 <dependency> 27 <groupId>org.springframework.boot</groupId> 28 <artifactId>spring-boot-starter-actuator</artifactId> 29 </dependency> 30 <dependency> 31 <groupId>org.springframework.boot</groupId> 32 <artifactId>spring-boot-starter-test</artifactId> 33 <scope>test</scope> 34 </dependency> 35 <!-- Spring Boot End --> 36 37 <!-- Spring Cloud Begin --> 38 <dependency> 39 <groupId>org.springframework.cloud</groupId> 40 <artifactId>spring-cloud-starter-stream-rocketmq</artifactId> 41 </dependency> 42 <!-- Spring Cloud End --> 43 </dependencies> 44 45 <build> 46 <plugins> 47 <plugin> 48 <groupId>org.springframework.boot</groupId> 49 <artifactId>spring-boot-maven-plugin</artifactId> 50 <configuration> 51 <mainClass>com.funtl.hello.spring.cloud.alibaba.rocketmq.provider.RocketMQProviderApplication</mainClass> 52 </configuration> 53 </plugin> 54 </plugins> 55 </build> 56 </project>
消息生产者服务
1 package com.funtl.hello.spring.cloud.alibaba.rocketmq.provider.service; 2 3 import org.springframework.beans.factory.annotation.Autowired; 4 import org.springframework.messaging.MessageChannel; 5 import org.springframework.messaging.support.MessageBuilder; 6 import org.springframework.stereotype.Service; 7 8 @Service 9 public class ProviderService { 10 @Autowired 11 private MessageChannel output; 12 13 public void send(String message) { 14 output.send(MessageBuilder.withPayload(message).build()); 15 } 16 }
Application
配置 Output(Source.class) 的 Binding 信息并配合 @EnableBinding 注解使其生效
1 package com.funtl.hello.spring.cloud.alibaba.rocketmq.provider; 2 3 import com.funtl.hello.spring.cloud.alibaba.rocketmq.provider.service.ProviderService; 4 import org.springframework.beans.factory.annotation.Autowired; 5 import org.springframework.boot.CommandLineRunner; 6 import org.springframework.boot.SpringApplication; 7 import org.springframework.boot.autoconfigure.SpringBootApplication; 8 import org.springframework.cloud.stream.annotation.EnableBinding; 9 import org.springframework.cloud.stream.messaging.Source; 10 11 @SpringBootApplication 12 @EnableBinding({Source.class}) 13 public class RocketMQProviderApplication implements CommandLineRunner { 14 15 @Autowired 16 private ProviderService providerService; 17 18 public static void main(String[] args) { 19 SpringApplication.run(RocketMQProviderApplication.class, args); 20 } 21 22 /** 23 * 实现了 CommandLineRunner 接口，只是为了 Spring Boot 启动时执行任务，没必要特别在乎 24 * @param args 25 * @throws Exception 26 */ 27 @Override 28 public void run(String... args) throws Exception { 29 providerService.send("Hello RocketMQ"); 30 } 31 }
application.yml
1 spring: 2 application: 3 name: rocketmq-provider 4 cloud: 5 stream: 6 rocketmq: 7 binder: 8 # RocketMQ 服务器地址 9 namesrv-addr: 192.168.10.149:9876 10 bindings: 11 # 这里是个 Map 类型参数，{} 为 YAML 中 Map 的行内写法 12 output: {destination: test-topic, content-type: application/json} 13 14 server: 15 port: 9093 16 17 management: 18 endpoints: 19 web: 20 exposure: 21 include: '*'
运行成功后便可在 RocketMQ 控制台的 消息 列表中选择 test-topic 主题便可看到发送的消息。
RocketMQ 消费者
POM
主要增长了 org.springframework.cloud:spring-cloud-starter-stream-rocketmq 依赖
1 <?xml version="1.0" encoding="UTF-8"?> 2 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 3 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4 <modelVersion>4.0.0</modelVersion> 5 6 <parent> 7 <groupId>com.funtl</groupId> 8 <artifactId>hello-spring-cloud-alibaba-dependencies</artifactId> 9 <version>1.0.0-SNAPSHOT</version> 10 <relativePath>../hello-spring-cloud-alibaba-dependencies/pom.xml</relativePath> 11 </parent> 12 13 <artifactId>hello-spring-cloud-alibaba-rocketmq-consumer</artifactId> 14 <packaging>jar</packaging> 15 16 <name>hello-spring-cloud-alibaba-rocketmq-consumer</name> 17 <url>http://www.funtl.com</url> 18 <inceptionYear>2018-Now</inceptionYear> 19 20 <dependencies> 21 <!-- Spring Boot Begin --> 22 <dependency> 23 <groupId>org.springframework.boot</groupId> 24 <artifactId>spring-boot-starter-web</artifactId> 25 </dependency> 26 <dependency> 27 <groupId>org.springframework.boot</groupId> 28 <artifactId>spring-boot-starter-actuator</artifactId> 29 </dependency> 30 <dependency> 31 <groupId>org.springframework.boot</groupId> 32 <artifactId>spring-boot-starter-test</artifactId> 33 <scope>test</scope> 34 </dependency> 35 <!-- Spring Boot End --> 36 37 <!-- Spring Cloud Begin --> 38 <dependency> 39 <groupId>org.springframework.cloud</groupId> 40 <artifactId>spring-cloud-starter-stream-rocketmq</artifactId> 41 </dependency> 42 <!-- Spring Cloud End --> 43 </dependencies> 44 45 <build> 46 <plugins> 47 <plugin> 48 <groupId>org.springframework.boot</groupId> 49 <artifactId>spring-boot-maven-plugin</artifactId> 50 <configuration> 51 <mainClass>com.funtl.hello.spring.cloud.alibaba.rocketmq.consumer.RocketMQConsumerApplication</mainClass> 52 </configuration> 53 </plugin> 54 </plugins> 55 </build> 56 </project>
消息消费者服务
主要使用 @StreamListener("input") 注解来订阅从名为 input 的 Binding 中接收的消息
1 package com.funtl.hello.spring.cloud.alibaba.rocketmq.consumer.receive; 2 3 import org.springframework.cloud.stream.annotation.StreamListener; 4 import org.springframework.stereotype.Service; 5 6 @Service 7 public class ConsumerReceive { 8 9 @StreamListener("input") 10 public void receiveInput(String message) { 11 System.out.println("Receive input: " + message); 12 } 13 }
Application
配置 Input(Sink.class) 的 Binding 信息并配合 @EnableBinding 注解使其生效
1 package com.funtl.hello.spring.cloud.alibaba.rocketmq.consumer; 2 3 import org.springframework.boot.SpringApplication; 4 import org.springframework.boot.autoconfigure.SpringBootApplication; 5 import org.springframework.cloud.stream.annotation.EnableBinding; 6 import org.springframework.cloud.stream.messaging.Sink; 7 8 @SpringBootApplication 9 @EnableBinding({Sink.class}) 10 public class RocketMQConsumerApplication { 11 public static void main(String[] args) { 12 SpringApplication.run(RocketMQConsumerApplication.class, args); 13 } 14 }
application.yml
1 spring: 2 application: 3 name: rocketmq-consumer 4 cloud: 5 stream: 6 rocketmq: 7 binder: 8 namesrv-addr: 192.168.10.149:9876 9 bindings: 10 input: {consumer.orderly: true} 11 bindings: 12 input: {destination: test-topic, content-type: text/plain, group: test-group, consumer.maxAttempts: 1} 13 14 server: 15 port: 9094 16 17 management: 18 endpoints: 19 web: 20 exposure: 21 include: '*'
运行成功后便可在控制台接收到消息：Receive input: Hello RocketMQ
RocketMQ 自定义 Binding
概述
在实际生产中，咱们须要发布和订阅的消息可能不止一种 Topic ，故此时就须要使用自定义 Binding 来帮咱们实现多 Topic 的发布和订阅功能
生产者
自定义 Output 接口，代码以下：
1 public interface MySource { 2 @Output("output1") 3 MessageChannel output1(); 4 5 @Output("output2") 6 MessageChannel output2(); 7 }
发布消息的案例代码以下：
1 @Autowired 2 private MySource source; 3 4 public void send(String msg) throws Exception { 5 source.output1().send(MessageBuilder.withPayload(msg).build()); 6 }
消费者
自定义 Input 接口，代码以下：
1 public interface MySink { 2 @Input("input1") 3 SubscribableChannel input1(); 4 5 @Input("input2") 6 SubscribableChannel input2(); 7 8 @Input("input3") 9 SubscribableChannel input3(); 10 11 @Input("input4") 12 SubscribableChannel input4(); 13 }
接收消息的案例代码以下：
1 @StreamListener("input1") 2 public void receiveInput1(String receiveMsg) { 3 System.out.println("input1 receive: " + receiveMsg); 4 }
Application
配置 Input 和 Output 的 Binding 信息并配合 @EnableBinding 注解使其生效，代码以下：
1 @SpringBootApplication 2 @EnableBinding({ MySource.class, MySink.class }) 3 public class RocketMQApplication { 4 public static void main(String[] args) { 5 SpringApplication.run(RocketMQApplication.class, args); 6 } 7 }
application.yml
生产者
1 spring: 2 application: 3 name: rocketmq-provider 4 cloud: 5 stream: 6 rocketmq: 7 binder: 8 namesrv-addr: 192.168.10.149:9876 9 bindings: 10 output1: {destination: test-topic1, content-type: application/json} 11 output2: {destination: test-topic2, content-type: application/json}
消费者
1 spring: 2 application: 3 name: rocketmq-consumer 4 cloud: 5 stream: 6 rocketmq: 7 binder: 8 namesrv-addr: 192.168.10.149:9876 9 bindings: 10 input: {consumer.orderly: true} 11 bindings: 12 input1: {destination: test-topic1, content-type: text/plain, group: test-group, consumer.maxAttempts: 1} 13 input2: {destination: test-topic1, content-type: text/plain, group: test-group, consumer.maxAttempts: 1} 14 input3: {destination: test-topic2, content-type: text/plain, group: test-group, consumer.maxAttempts: 1} 15 input4: {destination: test-topic2, content-type: text/plain, group: test-group, consumer.maxAttempts: 1}
引用：千峰
相关文章
- 1. RocketMQ分享
- 2. RocketMQ实战分享
- 3. 学习分享
- 4. RocketMQ 分享全纪实
- 5. RocketMq源码的分享
- 6. 【学习】025 RocketMQ
- 7. RocketMq学习
- 8. RocketMQ学习
- 9. Python学习分享
- 10. STM32学习分享
- 更多相关文章...
- • 您已经学习了 XML Schema，下一步学习什么呢？ - XML Schema 教程
- • 我们已经学习了 SQL，下一步学习什么呢？ - SQL 教程
- • Tomcat学习笔记（史上最全tomcat学习笔记）
- • 适用于PHP初学者的学习线路和建议
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. RocketMQ分享
- 2. RocketMQ实战分享
- 3. 学习分享
- 4. RocketMQ 分享全纪实
- 5. RocketMq源码的分享
- 6. 【学习】025 RocketMQ
- 7. RocketMq学习
- 8. RocketMQ学习
- 9. Python学习分享
- 10. STM32学习分享