mirror of
https://gitee.com/mirrors/Spring-Cloud-Alibaba.git
synced 2021-06-26 13:25:11 +08:00
add rocketmq binder example
This commit is contained in:
fangjian0423
parent
3263d203c2
commit
d0aa5dabbb
@ -27,6 +27,7 @@
|
||||
<module>ans-example/ans-consumer-ribbon-example</module>
|
||||
<module>ans-example/ans-provider-example</module>
|
||||
<module>acm-example/acm-local-example</module>
|
||||
<module>rocketmq-example</module>
|
||||
</modules>
|
||||
|
||||
<build>
|
||||
|
||||
51
spring-cloud-alibaba-examples/rocketmq-example/pom.xml
Normal file
51
spring-cloud-alibaba-examples/rocketmq-example/pom.xml
Normal file
@ -0,0 +1,51 @@
|
||||
<?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">
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-alibaba-examples</artifactId>
|
||||
<version>0.1.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
||||
<artifactId>rocketmq-example</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<description>Example demonstrating how to use rocketmq</description>
|
||||
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
|
||||
</dependency>
|
||||
<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>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-deploy-plugin</artifactId>
|
||||
<version>${maven-deploy-plugin.version}</version>
|
||||
<configuration>
|
||||
<skip>true</skip>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
|
||||
259
spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md
Normal file
259
spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md
Normal file
@ -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的架构设计。
|
||||
|
||||
|
||||
|
||||
## 示例
|
||||
|
||||
### 如何接入
|
||||
|
||||
在启动示例进行演示之前，我们先了解一下 Spring Cloud 应用如何接入 RocketMQ Binder。
|
||||
|
||||
> **注意：本章节只是为了便于您理解接入方式，本示例代码中已经完成****接入工作，您无需再进行修改。**
|
||||
|
||||
1. 首先，修改 `pom.xml` 文件，引入 RocketMQ Stream Starter。
|
||||
|
||||
```xml
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
|
||||
</dependency>
|
||||
```
|
||||
|
||||
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<String, Object> 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 中或者通过其他社区渠道向我们提出。
|
||||
|
||||
@ -0,0 +1,39 @@
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
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 + '\'' + '}';
|
||||
}
|
||||
}
|
||||
@ -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 <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@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);
|
||||
}
|
||||
|
||||
}
|
||||
@ -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 <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@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");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -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 <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@Service
|
||||
public class SenderService {
|
||||
|
||||
@Autowired
|
||||
private Source source;
|
||||
|
||||
public void send(String msg) throws Exception {
|
||||
source.output().send(MessageBuilder.withPayload(msg).build());
|
||||
}
|
||||
|
||||
public <T> void sendWithTags(T msg, String tag) throws Exception {
|
||||
Map<String, Object> map = new HashMap<>();
|
||||
map.put(MessageConst.PROPERTY_TAGS, tag);
|
||||
Message message = MessageBuilder.createMessage(msg, new MessageHeaders(map));
|
||||
source.output().send(message);
|
||||
}
|
||||
|
||||
public <T> 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);
|
||||
}
|
||||
|
||||
}
|
||||
@ -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=*
|
||||
Loading…
x
Reference in New Issue
Block a user