diff --git a/spring-cloud-alibaba-examples/pom.xml b/spring-cloud-alibaba-examples/pom.xml index 47a34f4f..98b392f8 100644 --- a/spring-cloud-alibaba-examples/pom.xml +++ b/spring-cloud-alibaba-examples/pom.xml @@ -27,6 +27,7 @@ ans-example/ans-consumer-ribbon-example ans-example/ans-provider-example acm-example/acm-local-example + rocketmq-example diff --git a/spring-cloud-alibaba-examples/rocketmq-example/pom.xml b/spring-cloud-alibaba-examples/rocketmq-example/pom.xml new file mode 100644 index 00000000..e8800187 --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/pom.xml @@ -0,0 +1,51 @@ + + + + + org.springframework.cloud + spring-cloud-alibaba-examples + 0.1.1.BUILD-SNAPSHOT + + 4.0.0 + + + rocketmq-example + jar + Example demonstrating how to use rocketmq + + + + + org.springframework.cloud + spring-cloud-starter-stream-rocketmq + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-starter-actuator + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-deploy-plugin + ${maven-deploy-plugin.version} + + true + + + + + + + diff --git a/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md b/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md new file mode 100644 index 00000000..b00bfbc3 --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md @@ -0,0 +1,259 @@ +# RocketMQ Example + +## 项目说明 + +本项目演示如何使用 RocketMQ Binder 完成 Spring Cloud 应用消息的订阅和发布。 + +[RocketMQ](https://rocketmq.apache.org/) 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。 + +在说明RocketMQ的示例之前，我们先了解一下 Spring Cloud Stream 中的Binder和Binding概念。 + +Binder: 跟外部消息中间件集成的组件，用来创建Binding，各消息中间件都有自己的Binder实现。 + +比如 `Kafka` 的实现 `KafkaMessageChannelBinder` ，`RabbitMQ` 的实现 `RabbitMessageChannelBinder` 以及 `RocketMQ` 的实现 `RocketMQMessageChannelBinder` 。 + +Binding: 包括Input Binding和Output Binding。 + +Binding在消息中间件与应用程序提供的Provider和Consumer之间提供了一个桥梁，实现了开发者只需使用应用程序的Provider或Consumer生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。 + +下图是Spring Cloud Stream的架构设计。 + +![](https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/images/SCSt-overview.png) + +## 示例 + +### 如何接入 + +在启动示例进行演示之前，我们先了解一下 Spring Cloud 应用如何接入 RocketMQ Binder。 + +> **注意：本章节只是为了便于您理解接入方式，本示例代码中已经完成****接入工作，您无需再进行修改。** + +1. 首先，修改 `pom.xml` 文件，引入 RocketMQ Stream Starter。 + +```xml + + org.springframework.cloud + spring-cloud-starter-stream-rocketmq + +``` + +2. 配置Input和Output的Binding信息并配合`@EnableBinding`注解使其生效 + +```java +@SpringBootApplication +@EnableBinding({ Source.class, Sink.class }) +public class RocketMQApplication { + public static void main(String[] args) { + SpringApplication.run(RocketMQApplication.class, args); + } +} +``` + +配置Binding信息： +```properties +# 配置rocketmq的nameserver地址 +spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876 +# 定义name为output的binding +spring.cloud.stream.bindings.output.destination=test-topic +spring.cloud.stream.bindings.output.content-type=application/json +# 定义name为input的binding +spring.cloud.stream.bindings.input.destination=test-topic +spring.cloud.stream.bindings.input.content-type=application/json +spring.cloud.stream.bindings.input.group=test-group + +``` + +3. 消息发送及消息订阅 + +### 下载并启动 RocketMQ + +在接入RocketMQ Binder之前，首先需要启动RocketMQ的Name Server和Broker。 + +1. 下载[RocketMQ最新的二进制文件](https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip)，并解压 + +2. 启动 Name Server + +```bash +sh bin/mqnamesrv +``` + +3. 启动Broker + +```bash +sh bin/mqbroker -n localhost:9876 +``` + +4. 创建Topic: test-topic + +```bash +sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic +``` + +### 应用启动 + +1. 增加配置，在应用的 /src/main/resources/application.properties 中添加基本配置信息 + +```properties +spring.application.name=rocketmq-example +server.port=28081 +``` + +2. 启动应用，支持 IDE 直接启动和编译打包后启动。 + + 1. IDE直接启动：找到主类 `RocketMQApplication`，执行 main 方法启动应用。 + 2. 打包编译后启动：首先执行 `mvn clean package` 将工程编译打包，然后执行 `java -jar rocketmq-example.jar`启动应用。 + + +### 消息处理 + +使用name为output对应的binding发送消息到test-topic这个topic。 + +使用2个input binding订阅数据。 + +input1: 订阅topic为test-topic的消息，顺序消费所有消息(顺序消费的前提是所有消息都在一个MessageQueue中) +input2: 订阅topic为test-topic的消息，异步消费tags为tagStr的消息，Consumer端线程池个数为20 + +配置信息如下： + +```properties +spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876 + +spring.cloud.stream.bindings.output.destination=test-topic +spring.cloud.stream.bindings.output.content-type=application/json + +spring.cloud.stream.bindings.input1.destination=test-topic +spring.cloud.stream.bindings.input1.content-type=text/plain +spring.cloud.stream.bindings.input1.group=test-group1 +spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true + +spring.cloud.stream.bindings.input2.destination=test-topic +spring.cloud.stream.bindings.input2.content-type=text/plain +spring.cloud.stream.bindings.input2.group=test-group2 +spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false +spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr +spring.cloud.stream.bindings.input2.consumer.concurrency=20 + +``` + +#### 消息发送 + +使用MessageChannel进行消息发送： + +```java +public class ProducerRunner implements CommandLineRunner { + @Autowired + private MessageChannel output; // 获取name为output的binding + @Override + public void run(String... args) throws Exception { + Map headers = new HashMap<>(); + headers.put(MessageConst.PROPERTY_TAGS, "tagStr"); + Message message = MessageBuilder.createMessage(msg, new MessageHeaders(headers)); + output.send(message); + } +} +``` + +或者使用RocketMQ原生的API进行消息发送: + +```java +public class RocketMQProducer { + DefaultMQProducer producer = new DefaultMQProducer("producer_group"); + producer.setNamesrvAddr("127.0.0.1:9876"); + producer.start(); + + Message msg = new Message("test-topic", "tagStr", "message from rocketmq producer".getBytes()); + producer.send(msg); +} +``` + +#### 消息接收 + +使用`@StreamListener`注解接收消息： + +```java +@Service +public class ReceiveService { + + @StreamListener("input1") + public void receiveInput1(String receiveMsg) { + System.out.println("input1 receive: " + receiveMsg); + } + + @StreamListener("input2") + public void receiveInput2(String receiveMsg) { + System.out.println("input2 receive: " + receiveMsg); + } + +} +``` + +## Endpoint 信息查看 + +Spring Boot 应用支持通过 Endpoint 来暴露相关信息，RocketMQ Stream Starter 也支持这一点。 + +在使用之前需要在 Maven 中添加 `spring-boot-starter-actuator`依赖，并在配置中允许 Endpoints 的访问。 +* Spring Boot 1.x 中添加配置 `management.security.enabled=false` +* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*` + +Spring Boot 1.x 可以通过访问 http://127.0.0.1:28081/rocketmq_binder 来查看 RocketMQ Binder Endpoint 的信息。Spring Boot 2.x 可以通过访问 http://127.0.0.1:28081/actuator/rocketmq-binder 来访问。 + +这里会统计消息最后一次发送的数据，消息发送成功或失败的次数，消息消费成功或失败的次数等数据。 + +```json +{ + "runtime": { + "lastSend.timestamp": 1542786623915 + }, + "metrics": { + "scs-rocketmq.consumer.test-topic.totalConsumed": { + "count": 11 + }, + "scs-rocketmq.consumer.test-topic.totalConsumedFailures": { + "count": 0 + }, + "scs-rocketmq.producer.test-topic.totalSentFailures": { + "count": 0 + }, + "scs-rocketmq.consumer.test-topic.consumedPerSecond": { + "count": 11, + "fifteenMinuteRate": 0.012163847780107841, + "fiveMinuteRate": 0.03614605351360527, + "meanRate": 0.3493213353657594, + "oneMinuteRate": 0.17099243039490175 + }, + "scs-rocketmq.producer.test-topic.totalSent": { + "count": 5 + }, + "scs-rocketmq.producer.test-topic.sentPerSecond": { + "count": 5, + "fifteenMinuteRate": 0.005540151995103271, + "fiveMinuteRate": 0.01652854617838251, + "meanRate": 0.10697493212602836, + "oneMinuteRate": 0.07995558537067671 + }, + "scs-rocketmq.producer.test-topic.sentFailuresPerSecond": { + "count": 0, + "fifteenMinuteRate": 0.0, + "fiveMinuteRate": 0.0, + "meanRate": 0.0, + "oneMinuteRate": 0.0 + }, + "scs-rocketmq.consumer.test-topic.consumedFailuresPerSecond": { + "count": 0, + "fifteenMinuteRate": 0.0, + "fiveMinuteRate": 0.0, + "meanRate": 0.0, + "oneMinuteRate": 0.0 + } + } +} +``` + +## More + +RocketMQ 是一款功能强大的分布式消息系统，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。 + +此 Demo 仅演示了 RocketMQ 与 Spring Cloud Stream 结合后的使用，更多 RocketMQ 相关的信息，请参考 [RocketMQ 项目](https://github.com/apache/rocketmq)。 + +如果您对 spring cloud starter stream rocketmq 有任何建议或想法，欢迎在 issue 中或者通过其他社区渠道向我们提出。 + diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java new file mode 100644 index 00000000..e98b6a10 --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java @@ -0,0 +1,39 @@ +package org.springframework.cloud.alibaba.cloud.examples; + +/** + * @author Jim + */ +public class Foo { + + private int id; + private String bar; + + public Foo() { + } + + public Foo(int id, String bar) { + this.id = id; + this.bar = bar; + } + + public int getId() { + return id; + } + + public void setId(int id) { + this.id = id; + } + + public String getBar() { + return bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + + @Override + public String toString() { + return "Foo{" + "id=" + id + ", bar='" + bar + '\'' + '}'; + } +} diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java new file mode 100644 index 00000000..f251902d --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java @@ -0,0 +1,33 @@ +package org.springframework.cloud.alibaba.cloud.examples; + +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.stereotype.Service; + +/** + * @author Jim + */ +@Service +public class ReceiveService { + + @StreamListener("input1") + public void receiveInput1(String receiveMsg) { + System.out.println("input1 receive: " + receiveMsg); + } + + @StreamListener("input2") + public void receiveInput2(String receiveMsg) { + System.out.println("input2 receive: " + receiveMsg); + } + + @StreamListener("input3") + public void receiveInput3(@Payload Foo foo) { + System.out.println("input3 receive: " + foo); + } + + @StreamListener("input1") + public void receiveInput1Again(String receiveMsg) { + System.out.println("input1 receive again: " + receiveMsg); + } + +} diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQApplication.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQApplication.java new file mode 100644 index 00000000..e4a8b1ab --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQApplication.java @@ -0,0 +1,65 @@ +package org.springframework.cloud.alibaba.cloud.examples; + +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.alibaba.cloud.examples.RocketMQApplication.MySink; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.SubscribableChannel; + +/** + * @author Jim + */ +@SpringBootApplication +@EnableBinding({ Source.class, MySink.class }) +public class RocketMQApplication { + + public interface MySink { + + @Input("input1") + SubscribableChannel input1(); + + @Input("input2") + SubscribableChannel input2(); + + @Input("input3") + SubscribableChannel input3(); + + } + + public static void main(String[] args) { + SpringApplication.run(RocketMQApplication.class, args); + } + + @Bean + public CustomRunner customRunner() { + return new CustomRunner(); + } + + public static class CustomRunner implements CommandLineRunner { + @Autowired + private SenderService senderService; + + @Override + public void run(String... args) throws Exception { + int count = 5; + for (int index = 1; index <= count; index++) { + String msgContent = "msg-" + index; + if (index % 3 == 0) { + senderService.send(msgContent); + } + else if (index % 3 == 1) { + senderService.sendWithTags(msgContent, "tagStr"); + } + else { + senderService.sendObject(new Foo(index, "foo"), "tagObj"); + } + } + } + } + +} diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java new file mode 100644 index 00000000..1c886d01 --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java @@ -0,0 +1,43 @@ +package org.springframework.cloud.alibaba.cloud.examples; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.rocketmq.common.message.MessageConst; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.stereotype.Service; +import org.springframework.util.MimeTypeUtils; + +/** + * @author Jim + */ +@Service +public class SenderService { + + @Autowired + private Source source; + + public void send(String msg) throws Exception { + source.output().send(MessageBuilder.withPayload(msg).build()); + } + + public void sendWithTags(T msg, String tag) throws Exception { + Map map = new HashMap<>(); + map.put(MessageConst.PROPERTY_TAGS, tag); + Message message = MessageBuilder.createMessage(msg, new MessageHeaders(map)); + source.output().send(message); + } + + public void sendObject(T msg, String tag) throws Exception { + Message message = MessageBuilder.withPayload(msg) + .setHeader(MessageConst.PROPERTY_TAGS, tag) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) + .build(); + source.output().send(message); + } + +} diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties b/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties new file mode 100644 index 00000000..80ea2d99 --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties @@ -0,0 +1,31 @@ +spring.cloud.stream.default-binder=rocketmq + +spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876 + +spring.cloud.stream.bindings.output.destination=test-topic +spring.cloud.stream.bindings.output.content-type=application/json + +spring.cloud.stream.bindings.input1.destination=test-topic +spring.cloud.stream.bindings.input1.content-type=text/plain +spring.cloud.stream.bindings.input1.group=test-group1 +spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true +spring.cloud.stream.bindings.input1.consumer.maxAttempts=1 + +spring.cloud.stream.bindings.input2.destination=test-topic +spring.cloud.stream.bindings.input2.content-type=text/plain +spring.cloud.stream.bindings.input2.group=test-group2 +spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false +spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr +spring.cloud.stream.bindings.input2.consumer.concurrency=20 +spring.cloud.stream.bindings.input2.consumer.maxAttempts=1 + +spring.cloud.stream.bindings.input3.destination=test-topic +spring.cloud.stream.bindings.input3.content-type=application/json +spring.cloud.stream.bindings.input3.group=test-group3 +spring.cloud.stream.rocketmq.bindings.input3.consumer.tags=tagObj +spring.cloud.stream.bindings.input3.consumer.concurrency=20 +spring.cloud.stream.bindings.input3.consumer.maxAttempts=1 + +server.port=28081 + +management.endpoints.web.exposure.include=* \ No newline at end of file