[ISSUE #218] Fix spring scopeTarget will repeat consumer instance

rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/resources/application.properties

@@ -11,4 +11,4 @@ demo.rocketmq.topic.user=user-topic
 demo.rocketmq.tag=tagA
 
 # another nameserver different global
-demo.rocketmq.myNameServer=127.0.0.1:9876
+demo.rocketmq.myNameServer=dev1.host.70yi.ren:9876

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ExtProducerResetConfiguration.java

@@ -20,6 +20,7 @@
 import java.lang.reflect.Field;
 import java.util.Map;
 import java.util.Objects;
+
 import org.apache.rocketmq.acl.common.AclClientRPCHook;
 import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.exception.MQClientException;
@@ -30,6 +31,7 @@
 import org.apache.rocketmq.spring.annotation.ExtRocketMQTemplateConfiguration;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
+import org.apache.rocketmq.spring.support.SpringBeanUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.aop.framework.AopProxyUtils;
@@ -57,20 +59,20 @@ public class ExtProducerResetConfiguration implements ApplicationContextAware, S
     private RocketMQMessageConverter rocketMQMessageConverter;
 
     public ExtProducerResetConfiguration(RocketMQMessageConverter rocketMQMessageConverter,
-        StandardEnvironment environment, RocketMQProperties rocketMQProperties) {
+                                         StandardEnvironment environment, RocketMQProperties rocketMQProperties) {
         this.rocketMQMessageConverter = rocketMQMessageConverter;
         this.environment = environment;
         this.rocketMQProperties = rocketMQProperties;
     }
 
     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-        this.applicationContext = (ConfigurableApplicationContext)applicationContext;
+        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
     }
 
     @Override
     public void afterSingletonsInstantiated() {
-        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(ExtRocketMQTemplateConfiguration.class);
+        Map<String, Object> beans = SpringBeanUtil.getBeansWithAnnotation(this.applicationContext, ExtRocketMQTemplateConfiguration.class);
 
         if (Objects.nonNull(beans)) {
             beans.forEach(this::registerTemplate);
@@ -85,7 +87,7 @@ private void registerTemplate(String beanName, Object bean) {
         }
 
         ExtRocketMQTemplateConfiguration annotation = clazz.getAnnotation(ExtRocketMQTemplateConfiguration.class);
-        GenericApplicationContext genericApplicationContext = (GenericApplicationContext)applicationContext;
+        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
         validate(annotation, genericApplicationContext);
 
         DefaultMQProducer mqProducer = createProducer(annotation);
@@ -97,7 +99,7 @@ private void registerTemplate(String beanName, Object bean) {
             throw new BeanDefinitionValidationException(String.format("Failed to startup MQProducer for RocketMQTemplate {}",
                 beanName), e);
         }
-        RocketMQTemplate rocketMQTemplate = (RocketMQTemplate)bean;
+        RocketMQTemplate rocketMQTemplate = (RocketMQTemplate) bean;
         rocketMQTemplate.setProducer(mqProducer);
         rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
         log.info("Set real producer to :{} {}", beanName, annotation.value());
@@ -156,7 +158,7 @@ private DefaultMQProducer createProducer(ExtRocketMQTemplateConfiguration annota
     }
 
     private void validate(ExtRocketMQTemplateConfiguration annotation,
-        GenericApplicationContext genericApplicationContext) {
+                          GenericApplicationContext genericApplicationContext) {
         if (genericApplicationContext.isBeanNameInUse(annotation.value())) {
             throw new BeanDefinitionValidationException(String.format("Bean {} has been used in Spring Application Context, " +
                     "please check the @ExtRocketMQTemplateConfiguration",

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java

@@ -17,17 +17,14 @@
 
 package org.apache.rocketmq.spring.autoconfigure;
 
-import java.util.Collections;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
 import org.apache.rocketmq.client.AccessChannel;
 import org.apache.rocketmq.spring.annotation.ConsumeMode;
 import org.apache.rocketmq.spring.annotation.MessageModel;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
 import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
+import org.apache.rocketmq.spring.support.SpringBeanUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.aop.framework.AopProxyUtils;
@@ -42,6 +39,11 @@
 import org.springframework.core.env.StandardEnvironment;
 import org.springframework.util.StringUtils;
 
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+
 @Configuration
 public class ListenerContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {
     private final static Logger log = LoggerFactory.getLogger(ListenerContainerConfiguration.class);
@@ -57,20 +59,21 @@ public class ListenerContainerConfiguration implements ApplicationContextAware,
     private RocketMQMessageConverter rocketMQMessageConverter;
 
     public ListenerContainerConfiguration(RocketMQMessageConverter rocketMQMessageConverter,
-        StandardEnvironment environment, RocketMQProperties rocketMQProperties) {
+                                          StandardEnvironment environment, RocketMQProperties rocketMQProperties) {
         this.rocketMQMessageConverter = rocketMQMessageConverter;
         this.environment = environment;
         this.rocketMQProperties = rocketMQProperties;
     }
 
     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-        this.applicationContext = (ConfigurableApplicationContext)applicationContext;
+        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
     }
 
     @Override
     public void afterSingletonsInstantiated() {
-        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class);
+        Map<String, Object> beans = SpringBeanUtil.getBeansWithAnnotation(this.applicationContext, RocketMQMessageListener.
+            class);
 
         if (Objects.nonNull(beans)) {
             beans.forEach(this::registerContainer);
@@ -90,7 +93,7 @@ private void registerContainer(String beanName, Object bean) {
         String topic = this.environment.resolvePlaceholders(annotation.topic());
 
         boolean listenerEnabled =
-            (boolean)rocketMQProperties.getConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
+            (boolean) rocketMQProperties.getConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
                 .getOrDefault(topic, true);
 
         if (!listenerEnabled) {
@@ -103,7 +106,7 @@ private void registerContainer(String beanName, Object bean) {
 
         String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
             counter.incrementAndGet());
-        GenericApplicationContext genericApplicationContext = (GenericApplicationContext)applicationContext;
+        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
 
         genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
             () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
@@ -122,11 +125,11 @@ private void registerContainer(String beanName, Object bean) {
     }
 
     private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean,
-        RocketMQMessageListener annotation) {
+                                                                             RocketMQMessageListener annotation) {
         DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();
-        
+
         container.setRocketMQMessageListener(annotation);
-        
+
         String nameServer = environment.resolvePlaceholders(annotation.nameServer());
         nameServer = StringUtils.isEmpty(nameServer) ? rocketMQProperties.getNameServer() : nameServer;
         String accessChannel = environment.resolvePlaceholders(annotation.accessChannel());
@@ -140,9 +143,18 @@ private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String
             container.setSelectorExpression(tags);
         }
         container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup()));
-        container.setRocketMQListener((RocketMQListener)bean);
+        container.setRocketMQListener((RocketMQListener) bean);
         container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
-        container.setName(name);  // REVIEW ME, use the same clientId or multiple?
+
+        /**
+         * REVIEW ME, use the same clientId or multiple?
+         *
+         * - this setting method is not work because on rocketmq-client( >= 4.6.0 ) will reset instanceName to pid
+         * - @see {org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#start}
+         * - @see {org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start}
+         * - @EinsiTang my-tangjianbin@163.com
+         */
+        container.setName(name);
 
         return container;
     }

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfiguration.java

@@ -119,6 +119,7 @@ public DefaultMQProducer defaultMQProducer(RocketMQProperties rocketMQProperties
         producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
         producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
         producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
+        producer.setInstanceName(rocketMQProperties.getClientInstanceName());
 
         return producer;
     }

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQProperties.java

@@ -20,6 +20,7 @@
 import org.apache.rocketmq.common.MixAll;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
+import java.net.InetAddress;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -37,6 +38,8 @@ public class RocketMQProperties {
      */
     private String accessChannel;
 
+    private String clientInstanceName = defaultClientInstanceName();
+
     private Producer producer;
 
     /**
@@ -73,6 +76,25 @@ public void setProducer(RocketMQProperties.Producer producer) {
         this.producer = producer;
     }
 
+    public String getClientInstanceName() {
+        return clientInstanceName;
+    }
+
+    public void setClientInstanceName(String clientInstanceName) {
+        this.clientInstanceName = clientInstanceName;
+    }
+
+    private String defaultClientInstanceName() {
+        String defaultName = null;
+        try {
+            defaultName = InetAddress.getLocalHost().getHostName();
+        } catch (Exception e) {
+            defaultName = "INSTANCE";
+        }
+
+        return defaultName;
+    }
+
     public static class Producer {
 
         /**

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQTransactionConfiguration.java

@@ -22,11 +22,13 @@
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+
 import org.apache.rocketmq.client.producer.TransactionMQProducer;
 import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
 import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.apache.rocketmq.spring.support.RocketMQUtil;
+import org.apache.rocketmq.spring.support.SpringBeanUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.aop.framework.AopProxyUtils;
@@ -44,12 +46,14 @@ public class RocketMQTransactionConfiguration implements ApplicationContextAware
 
     private ConfigurableApplicationContext applicationContext;
 
-    @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
         this.applicationContext = (ConfigurableApplicationContext) applicationContext;
     }
 
-    @Override public void afterSingletonsInstantiated() {
-        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQTransactionListener.class);
+    @Override
+    public void afterSingletonsInstantiated() {
+        Map<String, Object> beans = SpringBeanUtil.getBeansWithAnnotation(this.applicationContext, RocketMQTransactionListener.class);
 
         if (Objects.nonNull(beans)) {
             beans.forEach(this::registerTransactionListener);

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/SpringBeanUtil.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.rocketmq.spring.support;
+
+import org.springframework.aop.scope.ScopedProxyUtils;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.lang.NonNull;
+
+import java.lang.annotation.Annotation;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * override applicationContext.getBeansWithAnnotation method to make sure without same ProxyTarget bean
+ *
+ * author element
+ */
+public class SpringBeanUtil {
+
+    /**
+     * get beans with annotation
+     *
+     * @param applicationContext spring Application Context
+     * @param clazz              annotation class
+     * @return beans map without proxyTarget bean
+     */
+    public static Map<String, Object> getBeansWithAnnotation(@NonNull ConfigurableApplicationContext applicationContext, Class<? extends Annotation> clazz) {
+        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(clazz);
+        Map<String, Object> filterBeas = new HashMap<>(beans.size());
+        // remove proxy target
+        Set<Map.Entry<String, Object>> entrySet = beans.entrySet();
+        entrySet.forEach((entry) -> {
+            final String beanName = entry.getKey();
+            if (!ScopedProxyUtils.isScopedTarget(beanName)) {
+                filterBeas.put(beanName, entry.getValue());
+            }
+        });
+        return filterBeas;
+    }
+
+}