diff --git a/spring-cloud-stream-binder-rocketmq/pom.xml b/spring-cloud-stream-binder-rocketmq/pom.xml index b6c00c78..a4ad9a3e 100644 --- a/spring-cloud-stream-binder-rocketmq/pom.xml +++ b/spring-cloud-stream-binder-rocketmq/pom.xml @@ -56,12 +56,6 @@ rocketmq-spring-boot-starter - - io.micrometer - micrometer-core - true - - org.springframework.boot spring-boot-starter-test diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java index e864067c..152ccd31 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java @@ -17,9 +17,7 @@ package org.springframework.cloud.stream.binder.rocketmq; import java.util.HashMap; -import java.util.HashSet; import java.util.Map; -import java.util.Set; import org.apache.rocketmq.acl.common.AclClientRPCHook; import org.apache.rocketmq.acl.common.SessionCredentials; @@ -65,7 +63,6 @@ public class RocketMQMessageChannelBinder extends private final RocketMQProperties rocketMQProperties; private final InstrumentationManager instrumentationManager; - private Set clientConfigId = new HashSet<>(); private Map topicInUse = new HashMap<>(); public RocketMQMessageChannelBinder(RocketMQTopicProvisioner provisioningProvider, @@ -103,7 +100,6 @@ public class RocketMQMessageChannelBinder extends "there is more than 1 RocketMQTemplates in Spring BeanFactory"); } rocketMQTemplate = rocketMQTemplates.values().iterator().next(); - clientConfigId.add(rocketMQTemplate.getProducer().buildMQClientId()); } else { rocketMQTemplate = new RocketMQTemplate(); @@ -143,7 +139,6 @@ public class RocketMQMessageChannelBinder extends producer.setMaxMessageSize( producerProperties.getExtension().getMaxMessageSize()); rocketMQTemplate.setProducer(producer); - clientConfigId.add(producer.buildMQClientId()); } RocketMQMessageHandler messageHandler = new RocketMQMessageHandler( @@ -218,10 +213,6 @@ public class RocketMQMessageChannelBinder extends return extendedBindingProperties.getExtendedProducerProperties(channelName); } - public Set getClientConfigId() { - return clientConfigId; - } - public Map getTopicInUse() { return topicInUse; } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java index fe6ca85d..aa7ca366 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java @@ -19,13 +19,9 @@ package org.springframework.cloud.stream.binder.rocketmq.config; import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration; import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder; import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager; -import org.springframework.cloud.stream.binder.rocketmq.metrics.RocketMQBinderMetrics; import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties; import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties; import org.springframework.cloud.stream.binder.rocketmq.provisioning.RocketMQTopicProvisioner; @@ -33,9 +29,6 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; -import io.micrometer.core.instrument.MeterRegistry; -import io.micrometer.core.instrument.binder.MeterBinder; - /** * @author Timur Valiev * @author Jim @@ -84,21 +77,4 @@ public class RocketMQBinderAutoConfiguration { return new InstrumentationManager(); } - @Configuration - @ConditionalOnClass(MeterRegistry.class) - @ConditionalOnBean(MeterRegistry.class) - protected class RocketMQBinderMetricsConfiguration { - - @Bean - @ConditionalOnMissingBean(RocketMQBinderMetrics.class) - public MeterBinder rocketMqBinderMetrics( - RocketMQMessageChannelBinder rocketMQMessageChannelBinder, - RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties, - MeterRegistry meterRegistry) { - return new RocketMQBinderMetrics(rocketMQMessageChannelBinder, - rocketMQBinderConfigurationProperties, meterRegistry); - } - - } - } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/RocketMQListenerBindingContainer.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/RocketMQListenerBindingContainer.java index 0d2ba8bd..35f29ae8 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/RocketMQListenerBindingContainer.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/RocketMQListenerBindingContainer.java @@ -259,7 +259,6 @@ public class RocketMQListenerBindingContainer .prepareStart(consumer); } - rocketMQMessageChannelBinder.getClientConfigId().add(consumer.buildMQClientId()); } @Override diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/RocketMQBinderMetrics.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/RocketMQBinderMetrics.java deleted file mode 100644 index eef11f42..00000000 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/RocketMQBinderMetrics.java +++ /dev/null @@ -1,102 +0,0 @@ -/* - * Copyright (C) 2018 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.rocketmq.metrics; - -import java.util.Map; -import java.util.stream.Collectors; - -import org.apache.rocketmq.client.ClientConfig; -import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; -import org.apache.rocketmq.client.impl.MQClientManager; -import org.apache.rocketmq.client.producer.DefaultMQProducer; -import org.apache.rocketmq.common.message.MessageQueue; -import org.springframework.cloud.stream.binder.BindingCreatedEvent; -import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder; -import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties; -import org.springframework.context.ApplicationListener; - -import io.micrometer.core.instrument.Gauge; -import io.micrometer.core.instrument.MeterRegistry; -import io.micrometer.core.instrument.binder.MeterBinder; -import io.micrometer.core.lang.NonNull; - -/** - * @author Jim - */ -public class RocketMQBinderMetrics - implements MeterBinder, ApplicationListener { - - private final RocketMQMessageChannelBinder rocketMQMessageChannelBinder; - private final RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties; - private final MeterRegistry meterRegistry; - - static final String METRIC_NAME = "spring.cloud.stream.binder.rocketmq"; - - public RocketMQBinderMetrics( - RocketMQMessageChannelBinder rocketMQMessageChannelBinder, - RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties, - MeterRegistry meterRegistry) { - this.rocketMQMessageChannelBinder = rocketMQMessageChannelBinder; - this.rocketMQBinderConfigurationProperties = rocketMQBinderConfigurationProperties; - this.meterRegistry = meterRegistry; - } - - @Override - public void bindTo(@NonNull MeterRegistry registry) { - DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(); - pushConsumer - .setNamesrvAddr(rocketMQBinderConfigurationProperties.getNameServer()); - DefaultMQProducer producer = new DefaultMQProducer(); - producer.setNamesrvAddr(rocketMQBinderConfigurationProperties.getNameServer()); - - rocketMQMessageChannelBinder.getTopicInUse().forEach((topic, group) -> { - Gauge.builder(METRIC_NAME, this, o -> calculateMsgQueueOffset(topic, group)) - .tag("group", group).tag("topic", topic) - .description("RocketMQ all messageQueue size").register(registry); - }); - - } - - private double calculateMsgQueueOffset(String topic, String group) { - for (String clientConfigId : this.rocketMQMessageChannelBinder - .getClientConfigId()) { - ClientConfig clientConfig = new ClientConfig(); - String[] clientConfigArr = clientConfigId.split("@", 3); - clientConfig.setClientIP(clientConfigArr[0]); - clientConfig.setInstanceName(clientConfigArr[1]); - if (clientConfigArr.length > 2) { - clientConfig.setUnitName(clientConfigArr[2]); - } - Map queueLongMap = MQClientManager.getInstance() - .getAndCreateMQClientInstance(clientConfig) - .getConsumerStatus(topic, group); - if (queueLongMap.size() == 0) { - continue; - } - return queueLongMap.values().stream() - .collect(Collectors.summingLong(Long::longValue)); - } - return 0.0; - } - - @Override - public void onApplicationEvent(BindingCreatedEvent event) { - if (this.meterRegistry != null) { - this.bindTo(this.meterRegistry); - } - } -}