mirror of
https://gitee.com/mirrors/Spring-Cloud-Alibaba.git
synced 2021-06-26 13:25:11 +08:00
Merge pull request #129 from fangjian0423/1.x
Refactor binder and update examples for 1.x branch
This commit is contained in:
xiaojing
GitHub
commit
51e9ae552a
@ -25,7 +25,6 @@
|
||||
<alicloud.context.version>1.0.0</alicloud.context.version>
|
||||
<aliyun.sdk.edas.version>2.16.0</aliyun.sdk.edas.version>
|
||||
<rocketmq.version>4.3.1</rocketmq.version>
|
||||
<metrics.core>3.2.6</metrics.core>
|
||||
</properties>
|
||||
|
||||
<dependencyManagement>
|
||||
@ -221,13 +220,6 @@
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Third dependencies -->
|
||||
<dependency>
|
||||
<groupId>io.dropwizard.metrics</groupId>
|
||||
<artifactId>metrics-core</artifactId>
|
||||
<version>${metrics.core}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Testing Dependencies -->
|
||||
|
||||
|
||||
|
||||
@ -28,6 +28,10 @@
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.dropwizard.metrics</groupId>
|
||||
<artifactId>metrics-core</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
|
||||
@ -6,13 +6,21 @@
|
||||
|
||||
[RocketMQ](https://rocketmq.apache.org/) 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。
|
||||
|
||||
在说明RocketMQ的示例之前，我们先了解一下 Spring Cloud Stream 中的Binder和Binding概念。
|
||||
在说明RocketMQ的示例之前，我们先了解一下 Spring Cloud Stream。
|
||||
|
||||
Binder: 跟外部消息中间件集成的组件，用来创建Binding，各消息中间件都有自己的Binder实现。
|
||||
这是官方对Spring Cloud Stream的一段介绍：
|
||||
|
||||
Spring Cloud Stream是一个用于构建基于消息的微服务应用框架。它基于SpringBoot来创建具有生产级别的单机Spring应用，并且使用 `Spring Integration` 与Broker进行连接。
|
||||
|
||||
Spring Cloud Stream提供了消息中间件配置的统一抽象，推出了 publish-subscribe、consumer groups、partition 这些统一的概念。
|
||||
|
||||
Spring Cloud Stream 内部有2个概念：Binder 和 Binding。
|
||||
|
||||
* Binder: 跟外部消息中间件集成的组件，用来创建Binding，各消息中间件都有自己的Binder实现。
|
||||
|
||||
比如 `Kafka` 的实现 `KafkaMessageChannelBinder` ，`RabbitMQ` 的实现 `RabbitMessageChannelBinder` 以及 `RocketMQ` 的实现 `RocketMQMessageChannelBinder` 。
|
||||
|
||||
Binding: 包括Input Binding和Output Binding。
|
||||
* Binding: 包括Input Binding和Output Binding。
|
||||
|
||||
Binding在消息中间件与应用程序提供的Provider和Consumer之间提供了一个桥梁，实现了开发者只需使用应用程序的Provider或Consumer生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。
|
||||
|
||||
@ -110,8 +118,9 @@ server.port=28081
|
||||
|
||||
使用2个input binding订阅数据。
|
||||
|
||||
input1: 订阅topic为test-topic的消息，顺序消费所有消息(顺序消费的前提是所有消息都在一个MessageQueue中)
|
||||
input2: 订阅topic为test-topic的消息，异步消费tags为tagStr的消息，Consumer端线程池个数为20
|
||||
* input1：订阅topic为test-topic的消息，顺序消费所有消息(顺序消费的前提是所有消息都在一个MessageQueue中)
|
||||
|
||||
* input2：订阅topic为test-topic的消息，异步消费tags为tagStr的消息，Consumer端线程池个数为20
|
||||
|
||||
配置信息如下：
|
||||
|
||||
@ -249,6 +258,14 @@ Spring Boot 1.x 可以通过访问 http://127.0.0.1:28081/rocketmq_binder 来查
|
||||
}
|
||||
```
|
||||
|
||||
注意：要想查看统计数据需要在pom里加上 [metrics-core依赖](https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core) 。如若不加，endpoint将会显示warning信息而不会显示统计信息：
|
||||
|
||||
```json
|
||||
{
|
||||
"warning": "please add metrics-core dependency, we use it for metrics"
|
||||
}
|
||||
```
|
||||
|
||||
## More
|
||||
|
||||
RocketMQ 是一款功能强大的分布式消息系统，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。
|
||||
|
||||
@ -27,5 +27,4 @@ spring.cloud.stream.bindings.input3.consumer.concurrency=20
|
||||
spring.cloud.stream.bindings.input3.consumer.maxAttempts=1
|
||||
|
||||
server.port=28081
|
||||
|
||||
management.endpoints.web.exposure.include=*
|
||||
management.security.enabled=false
|
||||
@ -21,13 +21,15 @@
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.dropwizard.metrics</groupId>
|
||||
<artifactId>metrics-core</artifactId>
|
||||
<groupId>org.apache.rocketmq</groupId>
|
||||
<artifactId>rocketmq-client</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.rocketmq</groupId>
|
||||
<artifactId>rocketmq-client</artifactId>
|
||||
<groupId>io.dropwizard.metrics</groupId>
|
||||
<artifactId>metrics-core</artifactId>
|
||||
<scope>provided</scope>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
|
||||
@ -21,6 +21,8 @@ package org.springframework.cloud.stream.binder.rocketmq;
|
||||
*/
|
||||
public interface RocketMQBinderConstants {
|
||||
|
||||
String ENDPOINT_ID = "rocketmq_binder";
|
||||
|
||||
/**
|
||||
* Header key
|
||||
*/
|
||||
@ -33,10 +35,27 @@ public interface RocketMQBinderConstants {
|
||||
String ACKNOWLEDGEMENT_KEY = "ACKNOWLEDGEMENT";
|
||||
|
||||
/**
|
||||
* Instrumentation key
|
||||
* Instrumentation
|
||||
*/
|
||||
String LASTSEND_TIMESTAMP = "lastSend.timestamp";
|
||||
|
||||
String ENDPOINT_ID = "rocketmq_binder";
|
||||
interface Metrics {
|
||||
interface Producer {
|
||||
String PREFIX = "scs-rocketmq.producer.";
|
||||
String TOTAL_SENT = "totalSent";
|
||||
String TOTAL_SENT_FAILURES = "totalSentFailures";
|
||||
String SENT_PER_SECOND = "sentPerSecond";
|
||||
String SENT_FAILURES_PER_SECOND = "sentFailuresPerSecond";
|
||||
}
|
||||
|
||||
interface Consumer {
|
||||
String GROUP_PREFIX = "scs-rocketmq.consumerGroup.";
|
||||
String PREFIX = "scs-rocketmq.consumer.";
|
||||
String TOTAL_CONSUMED = "totalConsumed";
|
||||
String CONSUMED_PER_SECOND = "consumedPerSecond";
|
||||
String TOTAL_CONSUMED_FAILURES = "totalConsumedFailures";
|
||||
String CONSUMED_FAILURES_PER_SECOND = "consumedFailuresPerSecond";
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -42,76 +42,82 @@ import org.springframework.messaging.MessageHandler;
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQMessageChannelBinder extends
|
||||
AbstractMessageChannelBinder<ExtendedConsumerProperties<RocketMQConsumerProperties>,
|
||||
ExtendedProducerProperties<RocketMQProducerProperties>, RocketMQTopicProvisioner>
|
||||
implements ExtendedPropertiesBinder<MessageChannel, RocketMQConsumerProperties, RocketMQProducerProperties> {
|
||||
AbstractMessageChannelBinder<ExtendedConsumerProperties<RocketMQConsumerProperties>, ExtendedProducerProperties<RocketMQProducerProperties>, RocketMQTopicProvisioner>
|
||||
implements
|
||||
ExtendedPropertiesBinder<MessageChannel, RocketMQConsumerProperties, RocketMQProducerProperties> {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(RocketMQMessageChannelBinder.class);
|
||||
private static final Logger logger = LoggerFactory
|
||||
.getLogger(RocketMQMessageChannelBinder.class);
|
||||
|
||||
private final RocketMQExtendedBindingProperties extendedBindingProperties;
|
||||
private final RocketMQTopicProvisioner rocketTopicProvisioner;
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
private final ConsumersManager consumersManager;
|
||||
private final RocketMQExtendedBindingProperties extendedBindingProperties;
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
private final ConsumersManager consumersManager;
|
||||
|
||||
public RocketMQMessageChannelBinder(ConsumersManager consumersManager,
|
||||
RocketMQExtendedBindingProperties extendedBindingProperties,
|
||||
RocketMQTopicProvisioner provisioningProvider,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
|
||||
InstrumentationManager instrumentationManager) {
|
||||
super(true, null, provisioningProvider);
|
||||
this.consumersManager = consumersManager;
|
||||
this.extendedBindingProperties = extendedBindingProperties;
|
||||
this.rocketTopicProvisioner = provisioningProvider;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
public RocketMQMessageChannelBinder(ConsumersManager consumersManager,
|
||||
RocketMQExtendedBindingProperties extendedBindingProperties,
|
||||
RocketMQTopicProvisioner provisioningProvider,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
|
||||
InstrumentationManager instrumentationManager) {
|
||||
super(true, null, provisioningProvider);
|
||||
this.consumersManager = consumersManager;
|
||||
this.extendedBindingProperties = extendedBindingProperties;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
|
||||
ExtendedProducerProperties<RocketMQProducerProperties>
|
||||
producerProperties,
|
||||
MessageChannel errorChannel) throws Exception {
|
||||
if (producerProperties.getExtension().getEnabled()) {
|
||||
return new RocketMQMessageHandler(destination.getName(), producerProperties.getExtension(),
|
||||
rocketBinderConfigurationProperties, instrumentationManager);
|
||||
} else {
|
||||
throw new RuntimeException(
|
||||
"Binding for channel " + destination.getName() + "has been disabled, message can't be delivered");
|
||||
}
|
||||
}
|
||||
@Override
|
||||
protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
|
||||
ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
|
||||
MessageChannel errorChannel) throws Exception {
|
||||
if (producerProperties.getExtension().getEnabled()) {
|
||||
return new RocketMQMessageHandler(destination.getName(),
|
||||
producerProperties.getExtension(),
|
||||
rocketBinderConfigurationProperties, instrumentationManager);
|
||||
}
|
||||
else {
|
||||
throw new RuntimeException("Binding for channel " + destination.getName()
|
||||
+ " has been disabled, message can't be delivered");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties>
|
||||
consumerProperties)
|
||||
throws Exception {
|
||||
if (group == null || "".equals(group)) {
|
||||
throw new RuntimeException("'group' must be configured for channel + " + destination.getName());
|
||||
}
|
||||
@Override
|
||||
protected MessageProducer createConsumerEndpoint(ConsumerDestination destination,
|
||||
String group,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties)
|
||||
throws Exception {
|
||||
if (group == null || "".equals(group)) {
|
||||
throw new RuntimeException(
|
||||
"'group' must be configured for channel + " + destination.getName());
|
||||
}
|
||||
|
||||
RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(consumersManager,
|
||||
consumerProperties, destination.getName(), group, instrumentationManager);
|
||||
RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(
|
||||
consumersManager, consumerProperties, destination.getName(), group,
|
||||
instrumentationManager);
|
||||
|
||||
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, group,
|
||||
consumerProperties);
|
||||
if (consumerProperties.getMaxAttempts() > 1) {
|
||||
rocketInboundChannelAdapter.setRetryTemplate(buildRetryTemplate(consumerProperties));
|
||||
rocketInboundChannelAdapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
|
||||
} else {
|
||||
rocketInboundChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel());
|
||||
}
|
||||
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination,
|
||||
group, consumerProperties);
|
||||
if (consumerProperties.getMaxAttempts() > 1) {
|
||||
rocketInboundChannelAdapter
|
||||
.setRetryTemplate(buildRetryTemplate(consumerProperties));
|
||||
rocketInboundChannelAdapter
|
||||
.setRecoveryCallback(errorInfrastructure.getRecoverer());
|
||||
}
|
||||
else {
|
||||
rocketInboundChannelAdapter
|
||||
.setErrorChannel(errorInfrastructure.getErrorChannel());
|
||||
}
|
||||
|
||||
return rocketInboundChannelAdapter;
|
||||
}
|
||||
return rocketInboundChannelAdapter;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) {
|
||||
return extendedBindingProperties.getExtendedConsumerProperties(channelName);
|
||||
}
|
||||
@Override
|
||||
public RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) {
|
||||
return extendedBindingProperties.getExtendedConsumerProperties(channelName);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RocketMQProducerProperties getExtendedProducerProperties(String channelName) {
|
||||
return extendedBindingProperties.getExtendedProducerProperties(channelName);
|
||||
}
|
||||
@Override
|
||||
public RocketMQProducerProperties getExtendedProducerProperties(String channelName) {
|
||||
return extendedBindingProperties.getExtendedProducerProperties(channelName);
|
||||
}
|
||||
}
|
||||
|
||||
@ -20,11 +20,10 @@ import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderCon
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.actuate.endpoint.AbstractEndpoint;
|
||||
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
@ -32,8 +31,8 @@ import com.codahale.metrics.MetricRegistry;
|
||||
*/
|
||||
public class RocketMQBinderEndpoint extends AbstractEndpoint<Map<String, Object>> {
|
||||
|
||||
private MetricRegistry metricRegistry = new MetricRegistry();
|
||||
private Map<String, Object> runtime = new ConcurrentHashMap<>();
|
||||
@Autowired(required = false)
|
||||
private InstrumentationManager instrumentationManager;
|
||||
|
||||
public RocketMQBinderEndpoint() {
|
||||
super(ENDPOINT_ID);
|
||||
@ -42,17 +41,15 @@ public class RocketMQBinderEndpoint extends AbstractEndpoint<Map<String, Object>
|
||||
@Override
|
||||
public Map<String, Object> invoke() {
|
||||
Map<String, Object> result = new HashMap<>();
|
||||
result.put("metrics", metricRegistry().getMetrics());
|
||||
result.put("runtime", runtime());
|
||||
if (instrumentationManager != null) {
|
||||
result.put("metrics", instrumentationManager.getMetricRegistry());
|
||||
result.put("runtime", instrumentationManager.getRuntime());
|
||||
}
|
||||
else {
|
||||
result.put("warning",
|
||||
"please add metrics-core dependency, we use it for metrics");
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public MetricRegistry metricRegistry() {
|
||||
return metricRegistry;
|
||||
}
|
||||
|
||||
public Map<String, Object> runtime() {
|
||||
return runtime;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -16,6 +16,7 @@
|
||||
|
||||
package org.springframework.cloud.stream.binder.rocketmq.actuator;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.actuate.health.AbstractHealthIndicator;
|
||||
import org.springframework.boot.actuate.health.Health;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.Instrumentation;
|
||||
@ -27,40 +28,45 @@ import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationM
|
||||
*/
|
||||
public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
public RocketMQBinderHealthIndicator(InstrumentationManager instrumentationManager) {
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
@Autowired(required = false)
|
||||
private InstrumentationManager instrumentationManager;
|
||||
|
||||
@Override
|
||||
protected void doHealthCheck(Health.Builder builder) throws Exception {
|
||||
int upCount = 0, outOfServiceCount = 0;
|
||||
for (Instrumentation instrumentation : instrumentationManager
|
||||
.getHealthInstrumentations()) {
|
||||
if (instrumentation.isUp()) {
|
||||
upCount++;
|
||||
if (instrumentationManager != null) {
|
||||
for (Instrumentation instrumentation : instrumentationManager
|
||||
.getHealthInstrumentations()) {
|
||||
if (instrumentation.isUp()) {
|
||||
upCount++;
|
||||
}
|
||||
else if (instrumentation.isOutOfService()) {
|
||||
upCount++;
|
||||
}
|
||||
}
|
||||
else if (instrumentation.isOutOfService()) {
|
||||
upCount++;
|
||||
if (upCount == instrumentationManager.getHealthInstrumentations().size()) {
|
||||
builder.up();
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (upCount == instrumentationManager.getHealthInstrumentations().size()) {
|
||||
builder.up();
|
||||
return;
|
||||
}
|
||||
else if (outOfServiceCount == instrumentationManager.getHealthInstrumentations()
|
||||
.size()) {
|
||||
builder.outOfService();
|
||||
return;
|
||||
}
|
||||
builder.down();
|
||||
else if (outOfServiceCount == instrumentationManager
|
||||
.getHealthInstrumentations().size()) {
|
||||
builder.outOfService();
|
||||
return;
|
||||
}
|
||||
builder.down();
|
||||
|
||||
for (Instrumentation instrumentation : instrumentationManager
|
||||
.getHealthInstrumentations()) {
|
||||
if (!instrumentation.isStarted()) {
|
||||
builder.withException(instrumentation.getStartException());
|
||||
for (Instrumentation instrumentation : instrumentationManager
|
||||
.getHealthInstrumentations()) {
|
||||
if (!instrumentation.isStarted()) {
|
||||
builder.withException(instrumentation.getStartException());
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
builder.down();
|
||||
builder.withDetail("warning",
|
||||
"please add metrics-core dependency, we use it for metrics");
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@ -33,38 +33,46 @@ import org.springframework.context.annotation.Configuration;
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({RocketMQBinderConfigurationProperties.class, RocketMQExtendedBindingProperties.class})
|
||||
@EnableConfigurationProperties({ RocketMQBinderConfigurationProperties.class,
|
||||
RocketMQExtendedBindingProperties.class })
|
||||
public class RocketMQBinderAutoConfiguration {
|
||||
|
||||
private final RocketMQExtendedBindingProperties extendedBindingProperties;
|
||||
private final RocketMQExtendedBindingProperties extendedBindingProperties;
|
||||
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
|
||||
@Autowired
|
||||
public RocketMQBinderAutoConfiguration(RocketMQExtendedBindingProperties extendedBindingProperties,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
|
||||
this.extendedBindingProperties = extendedBindingProperties;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
System.setProperty(ClientLogger.CLIENT_LOG_LEVEL, this.rocketBinderConfigurationProperties.getLogLevel());
|
||||
}
|
||||
@Autowired(required = false)
|
||||
private InstrumentationManager instrumentationManager;
|
||||
|
||||
@Bean
|
||||
public RocketMQTopicProvisioner provisioningProvider() {
|
||||
return new RocketMQTopicProvisioner();
|
||||
}
|
||||
@Autowired
|
||||
public RocketMQBinderAutoConfiguration(
|
||||
RocketMQExtendedBindingProperties extendedBindingProperties,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
|
||||
this.extendedBindingProperties = extendedBindingProperties;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
System.setProperty(ClientLogger.CLIENT_LOG_LEVEL,
|
||||
this.rocketBinderConfigurationProperties.getLogLevel());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RocketMQMessageChannelBinder rocketMessageChannelBinder(RocketMQTopicProvisioner provisioningProvider,
|
||||
InstrumentationManager instrumentationManager,
|
||||
ConsumersManager consumersManager) {
|
||||
RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(consumersManager, extendedBindingProperties,
|
||||
provisioningProvider, rocketBinderConfigurationProperties, instrumentationManager);
|
||||
return binder;
|
||||
}
|
||||
@Bean
|
||||
public RocketMQTopicProvisioner provisioningProvider() {
|
||||
return new RocketMQTopicProvisioner();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ConsumersManager consumersManager(InstrumentationManager instrumentationManager) {
|
||||
return new ConsumersManager(instrumentationManager, rocketBinderConfigurationProperties);
|
||||
}
|
||||
@Bean
|
||||
public RocketMQMessageChannelBinder rocketMessageChannelBinder(
|
||||
RocketMQTopicProvisioner provisioningProvider,
|
||||
ConsumersManager consumersManager) {
|
||||
RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
|
||||
consumersManager, extendedBindingProperties, provisioningProvider,
|
||||
rocketBinderConfigurationProperties, instrumentationManager);
|
||||
return binder;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ConsumersManager consumersManager() {
|
||||
return new ConsumersManager(instrumentationManager,
|
||||
rocketBinderConfigurationProperties);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -17,7 +17,9 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.config;
|
||||
|
||||
import org.springframework.boot.actuate.autoconfigure.EndpointAutoConfiguration;
|
||||
import org.springframework.boot.actuate.endpoint.Endpoint;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderEndpoint;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderHealthIndicator;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
@ -29,6 +31,7 @@ import org.springframework.context.annotation.Configuration;
|
||||
*/
|
||||
@Configuration
|
||||
@AutoConfigureAfter(EndpointAutoConfiguration.class)
|
||||
@ConditionalOnClass(Endpoint.class)
|
||||
public class RocketMQBinderEndpointAutoConfiguration {
|
||||
|
||||
@Bean
|
||||
@ -37,16 +40,14 @@ public class RocketMQBinderEndpointAutoConfiguration {
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RocketMQBinderHealthIndicator rocketBinderHealthIndicator(
|
||||
InstrumentationManager instrumentationManager) {
|
||||
return new RocketMQBinderHealthIndicator(instrumentationManager);
|
||||
public RocketMQBinderHealthIndicator rocketBinderHealthIndicator() {
|
||||
return new RocketMQBinderHealthIndicator();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public InstrumentationManager instrumentationManager(
|
||||
RocketMQBinderEndpoint rocketBinderEndpoint) {
|
||||
return new InstrumentationManager(rocketBinderEndpoint.metricRegistry(),
|
||||
rocketBinderEndpoint.runtime());
|
||||
@ConditionalOnClass(name = "com.codahale.metrics.Counter")
|
||||
public InstrumentationManager instrumentationManager() {
|
||||
return new InstrumentationManager();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -38,85 +38,98 @@ import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsu
|
||||
*/
|
||||
public class ConsumersManager {
|
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(this.getClass());
|
||||
private final Logger logger = LoggerFactory.getLogger(this.getClass());
|
||||
|
||||
private final Map<String, DefaultMQPushConsumer> consumerGroups = new HashMap<>();
|
||||
private final Map<String, Boolean> started = new HashMap<>();
|
||||
private final Map<Map.Entry<String, String>, ExtendedConsumerProperties<RocketMQConsumerProperties>> propertiesMap
|
||||
= new HashMap<>();
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
private InstrumentationManager instrumentationManager;
|
||||
|
||||
public ConsumersManager(InstrumentationManager instrumentationManager,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
}
|
||||
private final Map<String, DefaultMQPushConsumer> consumerGroups = new HashMap<>();
|
||||
private final Map<String, Boolean> started = new HashMap<>();
|
||||
private final Map<Map.Entry<String, String>, ExtendedConsumerProperties<RocketMQConsumerProperties>> propertiesMap = new HashMap<>();
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
|
||||
public synchronized DefaultMQPushConsumer getOrCreateConsumer(String group, String topic,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) {
|
||||
propertiesMap.put(new AbstractMap.SimpleEntry<>(group, topic), consumerProperties);
|
||||
ConsumerGroupInstrumentation instrumentation = instrumentationManager.getConsumerGroupInstrumentation(group);
|
||||
instrumentationManager.addHealthInstrumentation(instrumentation);
|
||||
public ConsumersManager(InstrumentationManager instrumentationManager,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
}
|
||||
|
||||
if (consumerGroups.containsKey(group)) {
|
||||
return consumerGroups.get(group);
|
||||
}
|
||||
public synchronized DefaultMQPushConsumer getOrCreateConsumer(String group,
|
||||
String topic,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) {
|
||||
propertiesMap.put(new AbstractMap.SimpleEntry<>(group, topic),
|
||||
consumerProperties);
|
||||
if (instrumentationManager != null) {
|
||||
ConsumerGroupInstrumentation instrumentation = instrumentationManager
|
||||
.getConsumerGroupInstrumentation(group);
|
||||
instrumentationManager.addHealthInstrumentation(instrumentation);
|
||||
}
|
||||
|
||||
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
|
||||
consumer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
|
||||
consumerGroups.put(group, consumer);
|
||||
started.put(group, false);
|
||||
consumer.setConsumeThreadMax(consumerProperties.getConcurrency());
|
||||
consumer.setConsumeThreadMin(consumerProperties.getConcurrency());
|
||||
if (consumerProperties.getExtension().getBroadcasting()) {
|
||||
consumer.setMessageModel(MessageModel.BROADCASTING);
|
||||
}
|
||||
logger.info("RocketMQ consuming for SCS group {} created", group);
|
||||
return consumer;
|
||||
}
|
||||
if (consumerGroups.containsKey(group)) {
|
||||
return consumerGroups.get(group);
|
||||
}
|
||||
|
||||
public synchronized void startConsumers() throws MQClientException {
|
||||
for (String group : getConsumerGroups()) {
|
||||
start(group);
|
||||
}
|
||||
}
|
||||
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
|
||||
consumer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
|
||||
consumerGroups.put(group, consumer);
|
||||
started.put(group, false);
|
||||
consumer.setConsumeThreadMax(consumerProperties.getConcurrency());
|
||||
consumer.setConsumeThreadMin(consumerProperties.getConcurrency());
|
||||
if (consumerProperties.getExtension().getBroadcasting()) {
|
||||
consumer.setMessageModel(MessageModel.BROADCASTING);
|
||||
}
|
||||
logger.info("RocketMQ consuming for SCS group {} created", group);
|
||||
return consumer;
|
||||
}
|
||||
|
||||
public synchronized void startConsumer(String group) throws MQClientException {
|
||||
start(group);
|
||||
}
|
||||
public synchronized void startConsumers() throws MQClientException {
|
||||
for (String group : getConsumerGroups()) {
|
||||
start(group);
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void stopConsumer(String group) {
|
||||
stop(group);
|
||||
}
|
||||
public synchronized void startConsumer(String group) throws MQClientException {
|
||||
start(group);
|
||||
}
|
||||
|
||||
private void stop(String group) {
|
||||
if (consumerGroups.get(group) != null) {
|
||||
consumerGroups.get(group).shutdown();
|
||||
started.put(group, false);
|
||||
}
|
||||
}
|
||||
public synchronized void stopConsumer(String group) {
|
||||
stop(group);
|
||||
}
|
||||
|
||||
private synchronized void start(String group) throws MQClientException {
|
||||
if (started.get(group)) {
|
||||
return;
|
||||
}
|
||||
ConsumerGroupInstrumentation groupInstrumentation = instrumentationManager.getConsumerGroupInstrumentation(
|
||||
group);
|
||||
instrumentationManager.addHealthInstrumentation(groupInstrumentation);
|
||||
try {
|
||||
consumerGroups.get(group).start();
|
||||
started.put(group, true);
|
||||
groupInstrumentation.markStartedSuccessfully();
|
||||
} catch (MQClientException e) {
|
||||
groupInstrumentation.markStartFailed(e);
|
||||
logger.error("RocketMQ Consumer hasn't been started. Caused by " + e.getErrorMessage(), e);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
private void stop(String group) {
|
||||
if (consumerGroups.get(group) != null) {
|
||||
consumerGroups.get(group).shutdown();
|
||||
started.put(group, false);
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized Set<String> getConsumerGroups() {
|
||||
return consumerGroups.keySet();
|
||||
}
|
||||
private synchronized void start(String group) throws MQClientException {
|
||||
if (started.get(group)) {
|
||||
return;
|
||||
}
|
||||
ConsumerGroupInstrumentation groupInstrumentation = null;
|
||||
if (instrumentationManager != null) {
|
||||
groupInstrumentation = instrumentationManager
|
||||
.getConsumerGroupInstrumentation(group);
|
||||
instrumentationManager.addHealthInstrumentation(groupInstrumentation);
|
||||
}
|
||||
try {
|
||||
consumerGroups.get(group).start();
|
||||
started.put(group, true);
|
||||
if (groupInstrumentation != null) {
|
||||
groupInstrumentation.markStartedSuccessfully();
|
||||
}
|
||||
}
|
||||
catch (MQClientException e) {
|
||||
if (groupInstrumentation != null) {
|
||||
groupInstrumentation.markStartFailed(e);
|
||||
}
|
||||
logger.error("RocketMQ Consumer hasn't been started. Caused by "
|
||||
+ e.getErrorMessage(), e);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized Set<String> getConsumerGroups() {
|
||||
return consumerGroups.keySet();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -62,14 +62,14 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
|
||||
private ConsumerInstrumentation consumerInstrumentation;
|
||||
|
||||
private InstrumentationManager instrumentationManager;
|
||||
|
||||
private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
|
||||
|
||||
private final String destination;
|
||||
|
||||
private final String group;
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
private final ConsumersManager consumersManager;
|
||||
|
||||
private RetryTemplate retryTemplate;
|
||||
@ -89,21 +89,20 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
|
||||
@Override
|
||||
protected void doStart() {
|
||||
if (!consumerProperties.getExtension().getEnabled()) {
|
||||
if (consumerProperties == null
|
||||
|| !consumerProperties.getExtension().getEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
String tags = consumerProperties == null ? null
|
||||
: consumerProperties.getExtension().getTags();
|
||||
Boolean isOrderly = consumerProperties == null ? false
|
||||
: consumerProperties.getExtension().getOrderly();
|
||||
String tags = consumerProperties.getExtension().getTags();
|
||||
Boolean isOrderly = consumerProperties.getExtension().getOrderly();
|
||||
|
||||
DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group,
|
||||
destination, consumerProperties);
|
||||
|
||||
final CloudStreamMessageListener listener = isOrderly
|
||||
? new CloudStreamMessageListenerOrderly(instrumentationManager)
|
||||
: new CloudStreamMessageListenerConcurrently(instrumentationManager);
|
||||
? new CloudStreamMessageListenerOrderly()
|
||||
: new CloudStreamMessageListenerConcurrently();
|
||||
|
||||
if (retryTemplate != null) {
|
||||
retryTemplate.registerListener(listener);
|
||||
@ -116,9 +115,11 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
}
|
||||
}
|
||||
|
||||
consumerInstrumentation = instrumentationManager
|
||||
.getConsumerInstrumentation(destination);
|
||||
instrumentationManager.addHealthInstrumentation(consumerInstrumentation);
|
||||
if (instrumentationManager != null) {
|
||||
consumerInstrumentation = instrumentationManager
|
||||
.getConsumerInstrumentation(destination);
|
||||
instrumentationManager.addHealthInstrumentation(consumerInstrumentation);
|
||||
}
|
||||
|
||||
try {
|
||||
if (!StringUtils.isEmpty(consumerProperties.getExtension().getSql())) {
|
||||
@ -129,10 +130,14 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
consumer.subscribe(destination,
|
||||
org.apache.commons.lang3.StringUtils.join(tagsSet, " || "));
|
||||
}
|
||||
consumerInstrumentation.markStartedSuccessfully();
|
||||
if (consumerInstrumentation != null) {
|
||||
consumerInstrumentation.markStartedSuccessfully();
|
||||
}
|
||||
}
|
||||
catch (MQClientException e) {
|
||||
consumerInstrumentation.markStartFailed(e);
|
||||
if (consumerInstrumentation != null) {
|
||||
consumerInstrumentation.markStartFailed(e);
|
||||
}
|
||||
logger.error("RocketMQ Consumer hasn't been subscribed. Caused by "
|
||||
+ e.getErrorMessage(), e);
|
||||
throw new RuntimeException("RocketMQ Consumer hasn't been subscribed.", e);
|
||||
@ -166,19 +171,12 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
|
||||
protected class CloudStreamMessageListener implements MessageListener, RetryListener {
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
CloudStreamMessageListener(InstrumentationManager instrumentationManager) {
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
|
||||
Acknowledgement consumeMessage(final List<MessageExt> msgs) {
|
||||
boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null;
|
||||
try {
|
||||
if (enableRetry) {
|
||||
return RocketMQInboundChannelAdapter.this.retryTemplate.execute(
|
||||
// (RetryCallback<Acknowledgement, Exception>)
|
||||
new RetryCallback<Acknowledgement, Exception>() {
|
||||
return RocketMQInboundChannelAdapter.this.retryTemplate
|
||||
.execute(new RetryCallback<Acknowledgement, Exception>() {
|
||||
@Override
|
||||
public Acknowledgement doWithRetry(RetryContext context)
|
||||
throws Exception {
|
||||
@ -203,10 +201,12 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
}
|
||||
else {
|
||||
Acknowledgement result = doSendMsgs(msgs, null);
|
||||
instrumentationManager
|
||||
.getConsumerInstrumentation(
|
||||
RocketMQInboundChannelAdapter.this.destination)
|
||||
.markConsumed();
|
||||
if (RocketMQInboundChannelAdapter.this.instrumentationManager != null) {
|
||||
RocketMQInboundChannelAdapter.this.instrumentationManager
|
||||
.getConsumerInstrumentation(
|
||||
RocketMQInboundChannelAdapter.this.destination)
|
||||
.markConsumed();
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@ -214,10 +214,12 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
logger.error(
|
||||
"RocketMQ Message hasn't been processed successfully. Caused by ",
|
||||
e);
|
||||
instrumentationManager
|
||||
.getConsumerInstrumentation(
|
||||
RocketMQInboundChannelAdapter.this.destination)
|
||||
.markConsumedFailure();
|
||||
if (RocketMQInboundChannelAdapter.this.instrumentationManager != null) {
|
||||
RocketMQInboundChannelAdapter.this.instrumentationManager
|
||||
.getConsumerInstrumentation(
|
||||
RocketMQInboundChannelAdapter.this.destination)
|
||||
.markConsumedFailure();
|
||||
}
|
||||
throw new RuntimeException(
|
||||
"RocketMQ Message hasn't been processed successfully. Caused by ",
|
||||
e);
|
||||
@ -254,14 +256,17 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
@Override
|
||||
public <T, E extends Throwable> void close(RetryContext context,
|
||||
RetryCallback<T, E> callback, Throwable throwable) {
|
||||
if (RocketMQInboundChannelAdapter.this.instrumentationManager == null) {
|
||||
return;
|
||||
}
|
||||
if (throwable != null) {
|
||||
instrumentationManager
|
||||
RocketMQInboundChannelAdapter.this.instrumentationManager
|
||||
.getConsumerInstrumentation(
|
||||
RocketMQInboundChannelAdapter.this.destination)
|
||||
.markConsumedFailure();
|
||||
}
|
||||
else {
|
||||
instrumentationManager
|
||||
RocketMQInboundChannelAdapter.this.instrumentationManager
|
||||
.getConsumerInstrumentation(
|
||||
RocketMQInboundChannelAdapter.this.destination)
|
||||
.markConsumed();
|
||||
@ -277,11 +282,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
protected class CloudStreamMessageListenerConcurrently
|
||||
extends CloudStreamMessageListener implements MessageListenerConcurrently {
|
||||
|
||||
public CloudStreamMessageListenerConcurrently(
|
||||
InstrumentationManager instrumentationManager) {
|
||||
super(instrumentationManager);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs,
|
||||
ConsumeConcurrentlyContext context) {
|
||||
@ -295,11 +295,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
protected class CloudStreamMessageListenerOrderly extends CloudStreamMessageListener
|
||||
implements MessageListenerOrderly {
|
||||
|
||||
public CloudStreamMessageListenerOrderly(
|
||||
InstrumentationManager instrumentationManager) {
|
||||
super(instrumentationManager);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
|
||||
ConsumeOrderlyContext context) {
|
||||
|
||||
@ -45,14 +45,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
|
||||
|
||||
private ProducerInstrumentation producerInstrumentation;
|
||||
|
||||
private InstrumentationManager instrumentationManager;
|
||||
|
||||
private final RocketMQProducerProperties producerProperties;
|
||||
|
||||
private final String destination;
|
||||
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
protected volatile boolean running = false;
|
||||
|
||||
public RocketMQMessageHandler(String destination,
|
||||
@ -69,9 +69,11 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
|
||||
public void start() {
|
||||
producer = new DefaultMQProducer(destination);
|
||||
|
||||
producerInstrumentation = instrumentationManager
|
||||
.getProducerInstrumentation(destination);
|
||||
instrumentationManager.addHealthInstrumentation(producerInstrumentation);
|
||||
if (instrumentationManager != null) {
|
||||
producerInstrumentation = instrumentationManager
|
||||
.getProducerInstrumentation(destination);
|
||||
instrumentationManager.addHealthInstrumentation(producerInstrumentation);
|
||||
}
|
||||
|
||||
producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
|
||||
|
||||
@ -81,10 +83,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
|
||||
|
||||
try {
|
||||
producer.start();
|
||||
producerInstrumentation.markStartedSuccessfully();
|
||||
if (producerInstrumentation != null) {
|
||||
producerInstrumentation.markStartedSuccessfully();
|
||||
}
|
||||
}
|
||||
catch (MQClientException e) {
|
||||
producerInstrumentation.markStartFailed(e);
|
||||
if (producerInstrumentation != null) {
|
||||
producerInstrumentation.markStartFailed(e);
|
||||
}
|
||||
logger.error(
|
||||
"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
|
||||
throw new MessagingException(e.getMessage(), e);
|
||||
@ -142,14 +148,18 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
|
||||
RocketMQMessageHeaderAccessor.putSendResult((MutableMessage) message,
|
||||
sendRes);
|
||||
}
|
||||
instrumentationManager.getRuntime().put(
|
||||
RocketMQBinderConstants.LASTSEND_TIMESTAMP,
|
||||
System.currentTimeMillis());
|
||||
producerInstrumentation.markSent();
|
||||
if (instrumentationManager != null) {
|
||||
instrumentationManager.getRuntime().put(
|
||||
RocketMQBinderConstants.LASTSEND_TIMESTAMP,
|
||||
System.currentTimeMillis());
|
||||
producerInstrumentation.markSent();
|
||||
}
|
||||
}
|
||||
catch (MQClientException | RemotingException | MQBrokerException
|
||||
| InterruptedException | UnsupportedOperationException e) {
|
||||
producerInstrumentation.markSentFailure();
|
||||
if (producerInstrumentation != null) {
|
||||
producerInstrumentation.markSentFailure();
|
||||
}
|
||||
logger.error(
|
||||
"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
|
||||
throw new MessagingException(e.getMessage(), e);
|
||||
|
||||
@ -16,38 +16,44 @@
|
||||
|
||||
package org.springframework.cloud.stream.binder.rocketmq.metrics;
|
||||
|
||||
import static com.codahale.metrics.MetricRegistry.name;
|
||||
|
||||
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer;
|
||||
|
||||
import com.codahale.metrics.Counter;
|
||||
import com.codahale.metrics.Meter;
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
|
||||
import static com.codahale.metrics.MetricRegistry.name;
|
||||
|
||||
/**
|
||||
* @author juven.xuxb
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class ConsumerInstrumentation extends Instrumentation {
|
||||
|
||||
private final Counter totalConsumed;
|
||||
private final Counter totalConsumedFailures;
|
||||
private final Meter consumedPerSecond;
|
||||
private final Meter consumedFailuresPerSecond;
|
||||
private final Counter totalConsumed;
|
||||
private final Counter totalConsumedFailures;
|
||||
private final Meter consumedPerSecond;
|
||||
private final Meter consumedFailuresPerSecond;
|
||||
|
||||
public ConsumerInstrumentation(MetricRegistry registry, String baseMetricName) {
|
||||
super(baseMetricName);
|
||||
this.totalConsumed = registry.counter(name(baseMetricName, "totalConsumed"));
|
||||
this.consumedPerSecond = registry.meter(name(baseMetricName, "consumedPerSecond"));
|
||||
this.totalConsumedFailures = registry.counter(name(baseMetricName, "totalConsumedFailures"));
|
||||
this.consumedFailuresPerSecond = registry.meter(name(baseMetricName, "consumedFailuresPerSecond"));
|
||||
}
|
||||
public ConsumerInstrumentation(MetricRegistry registry, String baseMetricName) {
|
||||
super(baseMetricName);
|
||||
this.totalConsumed = registry
|
||||
.counter(name(baseMetricName, Consumer.TOTAL_CONSUMED));
|
||||
this.consumedPerSecond = registry
|
||||
.meter(name(baseMetricName, Consumer.CONSUMED_PER_SECOND));
|
||||
this.totalConsumedFailures = registry
|
||||
.counter(name(baseMetricName, Consumer.TOTAL_CONSUMED_FAILURES));
|
||||
this.consumedFailuresPerSecond = registry
|
||||
.meter(name(baseMetricName, Consumer.CONSUMED_FAILURES_PER_SECOND));
|
||||
}
|
||||
|
||||
public void markConsumed() {
|
||||
totalConsumed.inc();
|
||||
consumedPerSecond.mark();
|
||||
}
|
||||
public void markConsumed() {
|
||||
totalConsumed.inc();
|
||||
consumedPerSecond.mark();
|
||||
}
|
||||
|
||||
public void markConsumedFailure() {
|
||||
totalConsumedFailures.inc();
|
||||
consumedFailuresPerSecond.mark();
|
||||
}
|
||||
public void markConsumedFailure() {
|
||||
totalConsumedFailures.inc();
|
||||
consumedFailuresPerSecond.mark();
|
||||
}
|
||||
}
|
||||
|
||||
@ -22,28 +22,24 @@ import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Producer;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class InstrumentationManager {
|
||||
private final MetricRegistry metricRegistry;
|
||||
private final Map<String, Object> runtime;
|
||||
private final MetricRegistry metricRegistry = new MetricRegistry();
|
||||
private final Map<String, Object> runtime = new HashMap<>();
|
||||
private final Map<String, ProducerInstrumentation> producerInstrumentations = new HashMap<>();
|
||||
private final Map<String, ConsumerInstrumentation> consumeInstrumentations = new HashMap<>();
|
||||
private final Map<String, ConsumerGroupInstrumentation> consumerGroupsInstrumentations = new HashMap<>();
|
||||
|
||||
private final Map<String, Instrumentation> healthInstrumentations = new HashMap<>();
|
||||
|
||||
public InstrumentationManager(MetricRegistry metricRegistry,
|
||||
Map<String, Object> runtime) {
|
||||
this.metricRegistry = metricRegistry;
|
||||
this.runtime = runtime;
|
||||
}
|
||||
|
||||
public ProducerInstrumentation getProducerInstrumentation(String destination) {
|
||||
String key = "scs-rocketmq.producer." + destination;
|
||||
String key = Producer.PREFIX + destination;
|
||||
ProducerInstrumentation producerInstrumentation = producerInstrumentations
|
||||
.get(key);
|
||||
if (producerInstrumentation == null) {
|
||||
@ -54,7 +50,7 @@ public class InstrumentationManager {
|
||||
}
|
||||
|
||||
public ConsumerInstrumentation getConsumerInstrumentation(String destination) {
|
||||
String key = "scs-rocketmq.consumer." + destination;
|
||||
String key = Consumer.PREFIX + destination;
|
||||
ConsumerInstrumentation consumerInstrumentation = consumeInstrumentations
|
||||
.get(key);
|
||||
if (consumerInstrumentation == null) {
|
||||
@ -65,7 +61,7 @@ public class InstrumentationManager {
|
||||
}
|
||||
|
||||
public ConsumerGroupInstrumentation getConsumerGroupInstrumentation(String group) {
|
||||
String key = "scs-rocketmq.consumerGroup." + group;
|
||||
String key = Consumer.GROUP_PREFIX + group;
|
||||
ConsumerGroupInstrumentation consumerGroupInstrumentation = consumerGroupsInstrumentations
|
||||
.get(key);
|
||||
if (consumerGroupInstrumentation == null) {
|
||||
@ -86,4 +82,8 @@ public class InstrumentationManager {
|
||||
public Map<String, Object> getRuntime() {
|
||||
return runtime;
|
||||
}
|
||||
|
||||
public MetricRegistry getMetricRegistry() {
|
||||
return metricRegistry;
|
||||
}
|
||||
}
|
||||
|
||||
@ -16,38 +16,44 @@
|
||||
|
||||
package org.springframework.cloud.stream.binder.rocketmq.metrics;
|
||||
|
||||
import static com.codahale.metrics.MetricRegistry.name;
|
||||
|
||||
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Producer;
|
||||
|
||||
import com.codahale.metrics.Counter;
|
||||
import com.codahale.metrics.Meter;
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
|
||||
import static com.codahale.metrics.MetricRegistry.name;
|
||||
|
||||
/**
|
||||
* @author juven.xuxb
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class ProducerInstrumentation extends Instrumentation {
|
||||
|
||||
private final Counter totalSent;
|
||||
private final Counter totalSentFailures;
|
||||
private final Meter sentPerSecond;
|
||||
private final Meter sentFailuresPerSecond;
|
||||
private final Counter totalSent;
|
||||
private final Counter totalSentFailures;
|
||||
private final Meter sentPerSecond;
|
||||
private final Meter sentFailuresPerSecond;
|
||||
|
||||
public ProducerInstrumentation(MetricRegistry registry, String baseMetricName) {
|
||||
super(baseMetricName);
|
||||
this.totalSent = registry.counter(name(baseMetricName, "totalSent"));
|
||||
this.totalSentFailures = registry.counter(name(baseMetricName, "totalSentFailures"));
|
||||
this.sentPerSecond = registry.meter(name(baseMetricName, "sentPerSecond"));
|
||||
this.sentFailuresPerSecond = registry.meter(name(baseMetricName, "sentFailuresPerSecond"));
|
||||
}
|
||||
public ProducerInstrumentation(MetricRegistry registry, String baseMetricName) {
|
||||
super(baseMetricName);
|
||||
|
||||
public void markSent() {
|
||||
totalSent.inc();
|
||||
sentPerSecond.mark();
|
||||
}
|
||||
this.totalSent = registry.counter(name(baseMetricName, Producer.TOTAL_SENT));
|
||||
this.totalSentFailures = registry
|
||||
.counter(name(baseMetricName, Producer.TOTAL_SENT_FAILURES));
|
||||
this.sentPerSecond = registry
|
||||
.meter(name(baseMetricName, Producer.SENT_PER_SECOND));
|
||||
this.sentFailuresPerSecond = registry
|
||||
.meter(name(baseMetricName, Producer.SENT_FAILURES_PER_SECOND));
|
||||
}
|
||||
|
||||
public void markSentFailure() {
|
||||
totalSentFailures.inc();
|
||||
sentFailuresPerSecond.mark();
|
||||
}
|
||||
public void markSent() {
|
||||
totalSent.inc();
|
||||
sentPerSecond.mark();
|
||||
}
|
||||
|
||||
public void markSentFailure() {
|
||||
totalSentFailures.inc();
|
||||
sentFailuresPerSecond.mark();
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user