update docs

pull/142/head
fangjian0423 6 years ago
parent c665e97fb1
commit 25aecfe1ad

@ -1 +1,247 @@
== Spring Cloud Alibaba Rocket Binder
== Spring Cloud Alibaba RocketMQ Binder
### RocketMQ 介绍
https://rocketmq.apache.org[RocketMQ] 是一款开源的分布式消息系统,基于高可用分布式集群技术,提供低延时的、高可靠的消息发布与订阅服务。同时,广泛应用于多个领域,包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。
具有以下特点:
* 能够保证严格的消息顺序
* 提供丰富的消息拉取模式
* 高效的订阅者水平扩展能力
* 实时的消息订阅机制
* 亿级消息堆积能力
### RocketMQ 基本使用
* 下载 RocketMQ
下载 https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip[RocketMQ最新的二进制文件],并解压
解压后的目录结构如下:
```
apache-rocketmq
├── LICENSE
├── NOTICE
├── README.md
├── benchmark
├── bin
├── conf
└── lib
```
* 启动 NameServer
```bash
nohup sh bin/mqnamesrv &
tail -f ~/logs/rocketmqlogs/namesrv.log
```
* 启动 Broker
```bash
nohup sh bin/mqbroker -n localhost:9876 &
tail -f ~/logs/rocketmqlogs/broker.log
```
* 发送、接收消息
发送消息:
```bash
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
```
发送成功后显示:`SendResult [sendStatus=SEND_OK, msgId= ...`
接收消息:
```bash
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
```
接收成功后显示:`ConsumeMessageThread_%d Receive New Messages: [MessageExt...`
* 关闭 Server
```bash
sh bin/mqshutdown broker
sh bin/mqshutdown namesrv
```
### Spring Cloud Stream 介绍
Spring Cloud Stream 是一个用于构建基于消息的微服务应用框架。它基于 SpringBoot 来创建具有生产级别的单机 Spring 应用,并且使用 `Spring Integration` 与 Broker 进行连接。
Spring Cloud Stream 提供了消息中间件配置的统一抽象,推出了 publish-subscribe、consumer groups、partition 这些统一的概念。
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
image::https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/images/SCSt-overview.png[]
使用 Spring Cloud Stream 完成一段简单的消息发送和消息接收代码:
```java
MessageChannel messageChannel = new DirectChannel();
// 消息订阅
((SubscribableChannel) messageChannel).subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
System.out.println("receive msg: " + message.getPayload());
}
});
// 消息发送
messageChannel.send(MessageBuilder.withPayload("simple msg").build());
```
这段代码所有的消息类都是 `spring-messaging` 模块里提供的。屏蔽具体消息中间件的底层实现,如果想用更换消息中间件,在配置文件里配置相关消息中间件信息以及修改 binder 依赖即可。
**Spring Cloud Stream 底层也是基于这段代码去做了各种抽象。**
### Spring Cloud Alibaba RocketMQ Binder 实现原理
.RocketMQ Binder处理流程
image::https://cdn.nlark.com/lark/0/2018/png/64647/1543560843558-24525bf4-1d0e-4e10-be5f-bdde7127f6e6.png[]
RocketMQ Binder 的核心主要就是这3个类`RocketMQMessageChannelBinder``RocketMQInboundChannelAdapter` 和 `RocketMQMessageHandler`。
`RocketMQMessageChannelBinder` 是个标准的 Binder 实现,其内部构建 `RocketMQInboundChannelAdapter` 和 `RocketMQMessageHandler`。
`RocketMQMessageHandler` 用于 RocketMQ `Producer` 的启动以及消息的发送,其内部会根据 `spring-messaging` 模块内 `org.springframework.messaging.Message` 消息类,去创建 RocketMQ 的消息类 `org.apache.rocketmq.common.message.Message`。
在构造 `org.apache.rocketmq.common.message.Message` 的过程中会根据 `org.springframework.messaging.Message` 的 Header 构造成 `RocketMQMessageHeaderAccessor`。然后再根据 `RocketMQMessageHeaderAccessor` 中的一些属性,比如 tags、keys、flag等属性设置到 RocketMQ 的消息类 `org.apache.rocketmq.common.message.Message` 中。
`RocketMQInboundChannelAdapter` 用于 RocketMQ `Consumer` 的启动以及消息的接收。其内部还支持 https://github.com/spring-projects/spring-retry[spring-retry] 的使用。
在消费消息的时候可以从 Header 中获取 `Acknowledgement` 并进行一些设置。
比如使用 `MessageListenerConcurrently` 进行异步消费的时候,可以设置延迟消费:
```java
@StreamListener("input")
public void receive(Message message) {
RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message);
Acknowledgement acknowledgement = headerAccessor.getAcknowledgement(message);
acknowledgement.setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus.RECONSUME_LATER);
acknowledgement.setConsumeConcurrentlyDelayLevel(1);
}
```
比如使用 `MessageListenerOrderly` 进行顺序消费的时候,可以设置延迟消费:
```java
@StreamListener("input")
public void receive(Message message) {
RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message);
Acknowledgement acknowledgement = headerAccessor.getAcknowledgement(message);
acknowledgement.setConsumeOrderlyStatus(ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT);
acknowledgement.setConsumeOrderlySuspendCurrentQueueTimeMill(5000);
}
```
Provider端支持的配置
:frame: topbot
[width="60%",options="header"]
|====
^|配置项 ^|含义 ^| 默认值
|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.enabled`|是否启用producer|true
|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.max-message-size`|消息发送的最大字节数|0(大于0才会生效RocketMQ 默认值为4M = 1024 * 1024 * 4)
|====
Consumer端支持的配置
:frame: topbot
[width="60%",options="header"]
|====
^|配置项 ^|含义| 默认值
|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.enabled`|是否启用consumer|true
|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.tags`|Consumer订阅只有包括这些tags的topic消息。多个标签之间使用 "\|\|" 分割|
|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.sql`|Consumer订阅满足sql要求的topic消息|
|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.broadcasting`|Consumer是否是广播模式|false
|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.orderly`|顺序消费 or 异步消费|false
|====
### Endpoint支持
在使用Endpoint特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖,并在配置中允许 Endpoints 的访问。
* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的 endpoint 路径为 `/rocketmq_binder`
* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的 endpoint 路径为 `/actuator/rocketmq-binder`
Endpoint 会统计消息最后一次发送的数据,消息发送成功或失败的次数,消息消费成功或失败的次数等数据。
```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
}
}
}
```
注意要想查看统计数据需要在pom里加上 https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core[metrics-core依赖]。如若不加endpoint 将会显示 warning 信息而不会显示统计信息:
```json
{
"warning": "please add metrics-core dependency, we use it for metrics"
}
```

@ -1,20 +1,20 @@
== Spring Cloud Alibaba Sentinel
### Sentinel介绍
### Sentinel 介绍
随着微服务的流行服务和服务之间的稳定性变得越来越重要。Sentinel 以流量为切入点,从流量控制、熔断降级、系统负载保护等多个维度保护服务的稳定性。
随着微服务的流行,服务和服务之间的稳定性变得越来越重要。 https://github.com/alibaba/Sentinel[Sentinel] 以流量为切入点,从流量控制、熔断降级、系统负载保护等多个维度保护服务的稳定性。
Sentinel 具有以下特征:
https://github.com/alibaba/Sentinel[Sentinel] 具有以下特征:
* *丰富的应用场景*Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景,例如秒杀(即突发流量控制在系统容量可以承受的范围)、消息削峰填谷、实时熔断下游不可用应用等。
* *完备的实时监控*Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据,甚至 500 台以下规模的集群的汇总运行情况。
* *广泛的开源生态*Sentinel 提供开箱即用的与其它开源框架/库的整合模块,例如与 Spring Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。
* *完善的 SPI 扩展点*Sentinel 提供简单易用、完善的 SPI 扩展点。您可以通过实现扩展点,快速的定制逻辑。例如定制规则管理、适配数据源等。
* *丰富的应用场景* Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景,例如秒杀(即突发流量控制在系统容量可以承受的范围)、消息削峰填谷、实时熔断下游不可用应用等。
* *完备的实时监控* Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据,甚至 500 台以下规模的集群的汇总运行情况。
* *广泛的开源生态* Sentinel 提供开箱即用的与其它开源框架/库的整合模块,例如与 Spring Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。
* *完善的 SPI 扩展点* Sentinel 提供简单易用、完善的 SPI 扩展点。您可以通过实现扩展点,快速的定制逻辑。例如定制规则管理、适配数据源等。
### 如何使用Sentinel
### 如何使用 Sentinel
如果要在您的项目中引入Sentinel使用group ID为 `org.springframework.cloud` 和artifact ID为 `spring-cloud-starter-alibaba-sentinel` 的starter。
如果要在您的项目中引入 Sentinel使用 group ID 为 `org.springframework.cloud` 和 artifact ID 为 `spring-cloud-starter-alibaba-sentinel` 的 starter。
```xml
<dependency>
@ -23,7 +23,7 @@ Sentinel 具有以下特征:
</dependency>
```
下面这个例子就是一个最简单的使用Sentinel的例子:
下面这个例子就是一个最简单的使用 Sentinel 的例子:
```java
@SpringBootApplication
@ -47,9 +47,9 @@ public class TestController {
}
```
@SentinelResource注解用来标识资源是否被限流、降级。上述例子上该注解的属性'hello'表示资源名。
@SentinelResource 注解用来标识资源是否被限流、降级。上述例子上该注解的属性 'hello' 表示资源名。
@SentinelResource还提供了其它额外的属性如 `blockHandler``blockHandlerClass``fallback` 用于表示限流或降级的操作,更多内容可以参考 https://github.com/alibaba/Sentinel/wiki/%E6%B3%A8%E8%A7%A3%E6%94%AF%E6%8C%81[Sentinel注解支持]。
@SentinelResource 还提供了其它额外的属性如 `blockHandler``blockHandlerClass``fallback` 用于表示限流或降级的操作,更多内容可以参考 https://github.com/alibaba/Sentinel/wiki/%E6%B3%A8%E8%A7%A3%E6%94%AF%E6%8C%81[Sentinel注解支持]。
##### Sentinel 控制台
@ -74,7 +74,7 @@ image::https://github.com/alibaba/Sentinel/wiki/image/dashboard.png[]
###### 启动控制台
Sentinel 控制台是一个标准的SpringBoot应用以SpringBoot的方式运行jar包即可。
Sentinel 控制台是一个标准的 SpringBoot 应用,以 SpringBoot 的方式运行 jar 包即可。
```shell
java -Dserver.port=8080 -Dcsp.sentinel.dashboard.server=localhost:8080 -Dproject.name=sentinel-dashboard -jar sentinel-dashboard.jar
@ -94,17 +94,17 @@ spring:
dashboard: localhost:8080
----
这里的 `spring.cloud.sentinel.transport.port` 端口配置会在应用对应的机器上启动一个Http Server该Serve会与Sentinel控制台做交互。比如Sentinel控制台添加了1个限流规则会把规则数据push给这个Http Server接受Http Server再将规则注册到Sentinel中。
这里的 `spring.cloud.sentinel.transport.port` 端口配置会在应用对应的机器上启动一个 Http Server Server 会与 Sentinel 控制台做交互。比如 Sentinel 控制台添加了1个限流规则会把规则数据 push 给这个 Http Server 接收Http Server 再将规则注册到 Sentinel 中。
更多Sentinel控制台的使用及问题参考 https://github.com/alibaba/Sentinel/wiki/%E6%8E%A7%E5%88%B6%E5%8F%B0[Sentinel控制台]
更多 Sentinel 控制台的使用及问题参考: https://github.com/alibaba/Sentinel/wiki/%E6%8E%A7%E5%88%B6%E5%8F%B0[Sentinel控制台]
### Feign支持
### Feign 支持
DOING
### RestTemplate支持
### RestTemplate 支持
Spring Cloud Alibaba Sentinel支持对 `RestTemplate` 的服务调用使用Sentinel进行保护在构造 `RestTemplate` bean的时候需要加上 `@SentinelProtect` 注解。
Spring Cloud Alibaba Sentinel 支持对 `RestTemplate` 的服务调用使用 Sentinel 进行保护,在构造 `RestTemplate` bean的时候需要加上 `@SentinelProtect` 注解。
```java
@Bean
@ -122,12 +122,12 @@ public RestTemplate restTemplate() {
* `schema://host:port`:协议、主机和端口
NOTE: 以 `https://www.taobao.com/test` 这个url为例。对应的资源名有两种粒度分别是 `https://www.taobao.com:80` 以及 `https://www.taobao.com:80/test`
NOTE: 以 `https://www.taobao.com/test` 这个 url 为例。对应的资源名有两种粒度,分别是 `https://www.taobao.com:80` 以及 `https://www.taobao.com:80/test`
### 动态数据源支持
*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之前*要在Spring Cloud Alibaba Sentinel下使用动态数据源需要3个步骤
*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之前*,要在 Spring Cloud Alibaba Sentinel 下使用动态数据源需要3个步骤
* 配置文件中定义数据源信息。比如使用文件:
@ -141,7 +141,7 @@ spring.cloud.sentinel.datasource.converter=flowConverter
spring.cloud.sentinel.datasource.file=/Users/you/yourrule.json
----
* 创建一个Converter类实现 `com.alibaba.csp.sentinel.datasource.Converter` 接口,并且需要在 `ApplicationContext` 中有该类的一个Bean
* 创建一个 Converter 类实现 `com.alibaba.csp.sentinel.datasource.Converter` 接口,并且需要在 `ApplicationContext` 中有该类的一个 Bean
```java
@Component("flowConverter")
@ -154,7 +154,7 @@ public class JsonFlowRuleListParser implements Converter<String, List<FlowRule>>
}
```
这个Converter的bean name需要跟 `application.properties` 配置文件中的converter配置一致
这个 Converter bean name 需要跟 `application.properties` 配置文件中的 converter 配置一致
* 在任意一个 Spring Bean 中定义一个被 `@SentinelDataSource` 注解修饰的 `ReadableDataSource` 属性
@ -163,7 +163,7 @@ public class JsonFlowRuleListParser implements Converter<String, List<FlowRule>>
private ReadableDataSource dataSource;
```
`@SentinelDataSource` 注解的value属性表示数据源在 `application.properties` 配置文件中前缀。 该在例子中,前缀为 `spring.cloud.sentinel.datasource` 。
`@SentinelDataSource` 注解的 value 属性表示数据源在 `application.properties` 配置文件中前缀。 该在例子中,前缀为 `spring.cloud.sentinel.datasource` 。
如果 `ApplicationContext` 中存在超过1个 `ReadableDataSource` bean那么不会加载这些 `ReadableDataSource` 中的任意一个。 只有在 `ApplicationContext` 存在一个 `ReadableDataSource` 的情况下才会生效。
@ -173,7 +173,7 @@ private ReadableDataSource dataSource;
[Sentinel Starter] load 3 flow rules
```
*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之后*要在Spring Cloud Alibaba Sentinel下使用动态数据源只需要1个步骤
*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之后*,要在 Spring Cloud Alibaba Sentinel 下使用动态数据源只需要1个步骤
* 直接在 `application.properties` 配置文件中配置数据源信息即可
@ -196,23 +196,23 @@ spring.cloud.sentinel.datasource.ds4.apollo.default-flow-rule-value = test
```
这样配置方式参考了Spring Cloud Stream Binder的配置内部使用了 `TreeMap` 进行存储comparator为 `String.CASE_INSENSITIVE_ORDER` 。
这样配置方式参考了 Spring Cloud Stream Binder 的配置,内部使用了 `TreeMap` 进行存储comparator 为 `String.CASE_INSENSITIVE_ORDER` 。
NOTE: d1, ds2, ds3, ds4 是 `ReadableDataSource` 的名字,可随意编写。后面的 `file` `zk` `nacos` , `apollo` 就是对应具体的数据源。 它们后面的配置就是这些数据源各自的配置。
每种数据源都有两个共同的配置项: `data-type` 和 `converter-class` 。
`data-type` 配置项表示 `Converter`Spring Cloud Alibaba Sentinel默认提供两种内置的值分别是 `json` 和 `xml` (不填默认是json)。 如果不想使用内置的 `json` 或 `xml` 这两种 `Converter`,可以填写 `custom` 表示自定义 `Converter`,然后再配置 `converter-class` 配置项,该配置项需要写类的全路径名。
`data-type` 配置项表示 `Converter`Spring Cloud Alibaba Sentinel 默认提供两种内置的值,分别是 `json` 和 `xml` (不填默认是json)。 如果不想使用内置的 `json` 或 `xml` 这两种 `Converter`,可以填写 `custom` 表示自定义 `Converter`,然后再配置 `converter-class` 配置项,该配置项需要写类的全路径名。
这两种内置的 `Converter` 只支持解析 json数组 或 xml数组。内部解析的时候会自动判断每个json对象或xml对象属于哪4种Sentinel规则(`FlowRule``DegradeRule``SystemRule``AuthorityRule`)。
这两种内置的 `Converter` 只支持解析 json 数组 或 xml 数组。内部解析的时候会自动判断每个 json 对象或xml对象属于哪4种 Sentinel 规则(`FlowRule``DegradeRule``SystemRule``AuthorityRule`)。
比如10个规则数组里解析出5个限流规则和5个降级规则。 这种情况下该数据源不会注册,日志里页会进行警告。
如果10个规则里有9个限流规则1个解析报错了。这种情况下日志会警告有个规则格式错误另外9个限流规则会注册上去。
这里json或xml解析用的是 `jackson`。`ObjectMapper` 或 `XmlMapper` 使用默认的配置,遇到不认识的字段会解析报错。 因为不这样做的话限流 json 也会解析成 系统规则(系统规则所有的配置都有默认值),所以需要这样严格解析。
这里 json xml 解析用的是 `jackson`。`ObjectMapper` 或 `XmlMapper` 使用默认的配置,遇到不认识的字段会解析报错。 因为不这样做的话限流 json 也会解析成 系统规则(系统规则所有的配置都有默认值),所以需要这样严格解析。
当然还有一种情况是json对象或xml对象可能会匹配上所有4种规则(比如json对象里只配了 `resource` 字段那么会匹配上4种规则),这种情况下日志里会警告,并且过滤掉这个对象。
当然还有一种情况是 json 对象或 xml 对象可能会匹配上所有4种规则(比如json对象里只配了 `resource` 字段那么会匹配上4种规则),这种情况下日志里会警告,并且过滤掉这个对象。
用户使用这种配置的时候只需要填写正确的json或xml就行有任何不合理的信息都会在日志里打印出来。
@ -223,16 +223,16 @@ NOTE: d1, ds2, ds3, ds4 是 `ReadableDataSource` 的名字,可随意编写。
[Sentinel Starter] DataSource ds2-sentinel-nacos-datasource load 2 FlowRule
```
NOTE: 默认情况下xml格式是不支持的。需要添加 `jackson-dataformat-xml` 依赖后才会自动生效。
NOTE: 默认情况下xml 格式是不支持的。需要添加 `jackson-dataformat-xml` 依赖后才会自动生效。
关于Sentinel动态数据源的实现原理参考 https://github.com/alibaba/Sentinel/wiki/%E5%8A%A8%E6%80%81%E8%A7%84%E5%88%99%E6%89%A9%E5%B1%95[动态规则扩展]
关于 Sentinel 动态数据源的实现原理,参考: https://github.com/alibaba/Sentinel/wiki/%E5%8A%A8%E6%80%81%E8%A7%84%E5%88%99%E6%89%A9%E5%B1%95[动态规则扩展]
### Endpoint支持
### Endpoint 支持
在使用Endpoint特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖,并在配置中允许 Endpoints 的访问。
在使用 Endpoint 特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖,并在配置中允许 Endpoints 的访问。
* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的endpoint路径为 `/sentinel`
* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的endpoint路径为 `/actuator/sentinel`
* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的 endpoint 路径为 `/sentinel`
* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的 endpoint 路径为 `/actuator/sentinel`
### More

@ -6,17 +6,25 @@
[RocketMQ](https://rocketmq.apache.org/) 是一款开源的分布式消息系统,基于高可用分布式集群技术,提供低延时的、高可靠的消息发布与订阅服务。
在说明RocketMQ的示例之前我们先了解一下 Spring Cloud Stream 中的Binder和Binding概念
在说明 RocketMQ 的示例之前,我们先了解一下 Spring Cloud Stream。
Binder: 跟外部消息中间件集成的组件用来创建Binding各消息中间件都有自己的Binder实现。
这是官方对 Spring Cloud Stream 的一段介绍:
比如 `Kafka` 的实现 `KafkaMessageChannelBinder` `RabbitMQ` 的实现 `RabbitMessageChannelBinder` 以及 `RocketMQ` 的实现 `RocketMQMessageChannelBinder`
Spring Cloud Stream 是一个用于构建基于消息的微服务应用框架。它基于 SpringBoot 来创建具有生产级别的单机 Spring 应用,并且使用 `Spring Integration` 与 Broker 进行连接。
Spring Cloud Stream 提供了消息中间件配置的统一抽象,推出了 publish-subscribe、consumer groups、partition 这些统一的概念。
Binding: 包括Input Binding和Output Binding。
Spring Cloud Stream 内部有两个概念Binder 和 Binding。
Binding在消息中间件与应用程序提供的Provider和Consumer之间提供了一个桥梁实现了开发者只需使用应用程序的Provider或Consumer生产或消费数据即可屏蔽了开发者与底层消息中间件的接触
* Binder: 跟外部消息中间件集成的组件,用来创建 Binding各消息中间件都有自己的 Binder 实现
下图是Spring Cloud Stream的架构设计。
比如 `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)
@ -37,7 +45,7 @@ Binding在消息中间件与应用程序提供的Provider和Consumer之间提供
</dependency>
```
2. 配置Input和Output的Binding信息并配合`@EnableBinding`注解使其生效
2. 配置 Input Output Binding 信息并配合 `@EnableBinding` 注解使其生效
```java
@SpringBootApplication
@ -49,7 +57,7 @@ public class RocketMQApplication {
}
```
配置Binding信息
配置 Binding 信息:
```properties
# 配置rocketmq的nameserver地址
spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876
@ -67,7 +75,7 @@ spring.cloud.stream.bindings.input.group=test-group
### 下载并启动 RocketMQ
在接入RocketMQ Binder之前首先需要启动RocketMQ的Name Server和Broker。
在接入 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),并解压
@ -77,13 +85,13 @@ spring.cloud.stream.bindings.input.group=test-group
sh bin/mqnamesrv
```
3. 启动Broker
3. 启动 Broker
```bash
sh bin/mqbroker -n localhost:9876
```
4. 创建Topic: test-topic
4. 创建 Topic: test-topic
```bash
sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic
@ -100,18 +108,19 @@ server.port=28081
2. 启动应用,支持 IDE 直接启动和编译打包后启动。
1. IDE直接启动找到主类 `RocketMQApplication`,执行 main 方法启动应用。
2. 打包编译后启动:首先执行 `mvn clean package` 将工程编译打包,然后执行 `java -jar rocketmq-example.jar`启动应用。
1. IDE 直接启动:找到主类 `RocketMQApplication`,执行 main 方法启动应用。
2. 打包编译后启动:首先执行 `mvn clean package` 将工程编译打包,然后执行 `java -jar rocketmq-example.jar` 启动应用。
### 消息处理
使用name为output对应的binding发送消息到test-topic这个topic。
使用 name 为 output 对应的 binding 发送消息到 test-topic 这个 topic。
使用2个 input binding 订阅数据。
使用2个input binding订阅数据。
* input1: 订阅 topic 为 test-topic 的消息,顺序消费所有消息(顺序消费的前提是所有消息都在一个 MessageQueue 中)
input1: 订阅topic为test-topic的消息顺序消费所有消息(顺序消费的前提是所有消息都在一个MessageQueue中)
input2: 订阅topic为test-topic的消息异步消费tags为tagStr的消息Consumer端线程池个数为20
* input2: 订阅 topic 为 test-topic 的消息,异步消费 tags 为 tagStr 的消息Consumer 端线程池个数为20
配置信息如下:
@ -137,7 +146,7 @@ spring.cloud.stream.bindings.input2.consumer.concurrency=20
#### 消息发送
使用MessageChannel进行消息发送
使用 MessageChannel 进行消息发送:
```java
public class ProducerRunner implements CommandLineRunner {
@ -153,7 +162,7 @@ public class ProducerRunner implements CommandLineRunner {
}
```
或者使用RocketMQ原生的API进行消息发送:
或者使用 RocketMQ 原生的 API 进行消息发送:
```java
public class RocketMQProducer {
@ -168,7 +177,7 @@ public class RocketMQProducer {
#### 消息接收
使用`@StreamListener`注解接收消息:
使用 `@StreamListener` 注解接收消息:
```java
@Service
@ -249,11 +258,11 @@ Spring Boot 1.x 可以通过访问 http://127.0.0.1:18083/rocketmq_binder 来查
}
```
注意要想查看统计数据需要在pom里加上 [metrics-core依赖](https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core) 。如若不加endpoint将会显示warning信息而不会显示统计信息
注意要想查看统计数据需要在pom里加上 [metrics-core依赖](https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core)。如若不加endpoint 将会显示 warning 信息而不会显示统计信息:
```json
{
"warning": "please add metrics-core dependency, we use it to metrics"
"warning": "please add metrics-core dependency, we use it for metrics"
}
```

Loading…
Cancel
Save