From a2aff61fe3ccbdaa2745868c1334a579bec519f2 Mon Sep 17 00:00:00 2001 From: zhaowg Date: Fri, 23 Mar 2018 17:11:56 +0800 Subject: [PATCH] 整合阿里云RocketMQ SDK --- pom.xml | 11 ----------- src/main/java/org/apache/rocketmq/spring/starter/AliyunRocketMQAutoConfiguration.java | 36 ++++++++++++++++++++---------------- src/main/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfiguration.java | 197 ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- src/main/java/org/apache/rocketmq/spring/starter/RocketMQProperties.java | 19 +++++++++++-------- src/main/java/org/apache/rocketmq/spring/starter/annotation/RocketMQMessageListener.java | 10 ++++++---- src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQListenerContainer.java | 292 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------------------------------------------------------------------------------------------------------------------------------- src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQPushConsumerLifecycleListener.java | 23 +++++++++++++++++++++++ src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainer.java | 710 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainerConstants.java | 18 ++++++++++++++++-- src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQPushConsumerLifecycleListener.java | 23 ----------------------- src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQTemplate.java | 613 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ src/main/java/org/apache/rocketmq/spring/starter/enums/ConsumeMode.java | 11 ++++++++--- src/main/java/org/apache/rocketmq/spring/starter/enums/SelectorType.java | 2 +- src/main/java/org/apache/rocketmq/spring/starter/utils/ExceptionUtil.java | 21 +++++++++++++++++++++ src/test/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfigurationTests.java | 368 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- 15 files changed, 981 insertions(+), 1373 deletions(-) delete mode 100644 src/main/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfiguration.java create mode 100644 src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQPushConsumerLifecycleListener.java delete mode 100644 src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQPushConsumerLifecycleListener.java create mode 100644 src/main/java/org/apache/rocketmq/spring/starter/utils/ExceptionUtil.java diff --git a/pom.xml b/pom.xml index 19813e1..0792ff1 100644 --- a/pom.xml +++ b/pom.xml @@ -60,17 +60,6 @@ spring-boot-starter - org.apache.rocketmq - rocketmq-client - ${rocketmq-version} - - - org.slf4j - slf4j-api - - - - com.aliyun.openservices ons-client 1.7.2.Final diff --git a/src/main/java/org/apache/rocketmq/spring/starter/AliyunRocketMQAutoConfiguration.java b/src/main/java/org/apache/rocketmq/spring/starter/AliyunRocketMQAutoConfiguration.java index cdc48b9..cd1a2db 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/AliyunRocketMQAutoConfiguration.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/AliyunRocketMQAutoConfiguration.java @@ -22,13 +22,12 @@ import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerCon import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_CONSUME_MODE; import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_CONSUME_THREAD_MAX; import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_MESSAGE_MODEL; -import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_NAMESERVER; import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_OBJECT_MAPPER; import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_ROCKETMQ_LISTENER; import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_ROCKETMQ_TEMPLATE; import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_SELECTOR_EXPRESS; import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_SELECTOR_TYPE; -import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.PROP_TOPIC; +import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.*; import java.util.Map; import java.util.Objects; @@ -39,7 +38,6 @@ import javax.annotation.Resource; import org.apache.rocketmq.spring.starter.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.starter.core.AliyunRocketMQListenerContainer; -import org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainer; import org.apache.rocketmq.spring.starter.core.RocketMQListener; import org.apache.rocketmq.spring.starter.core.RocketMQTemplate; import org.springframework.aop.support.AopUtils; @@ -72,7 +70,6 @@ import lombok.extern.slf4j.Slf4j; @Configuration @EnableConfigurationProperties(RocketMQProperties.class) -@ConditionalOnProperty(name = "spring.rocketmq.aliyun",havingValue="true") @Order @Slf4j public class AliyunRocketMQAutoConfiguration { @@ -80,23 +77,27 @@ public class AliyunRocketMQAutoConfiguration { @Bean @ConditionalOnClass(Producer.class) @ConditionalOnMissingBean(Producer.class) - @ConditionalOnProperty(prefix = "spring.rocketmq", value = {"nameServer", "producer.group"}) + @ConditionalOnProperty(prefix = "spring.rocketmq", value = {"environmentPrefix", "producer.group"}) public Producer mqProducer(RocketMQProperties rocketMQProperties) { RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer(); String groupName = producerConfig.getGroup(); Assert.hasText(groupName, "[spring.rocketmq.producer.group] must not be null"); String accessKey = rocketMQProperties.getAccessKey(); - Assert.hasText(accessKey, "[spring.rocketmq.producer.accessKey] must not be null"); + Assert.hasText(accessKey, "[spring.rocketmq.accessKey] must not be null"); String secretKey = rocketMQProperties.getSecretKey(); - Assert.hasText(secretKey, "[spring.rocketmq.producer.secretKey] must not be null"); + Assert.hasText(secretKey, "[spring.rocketmq.secretKey] must not be null"); + String onsAddr = rocketMQProperties.getOnsAddr(); + Assert.hasText(secretKey, "[spring.rocketmq.onsAddr] must not be null"); + String environmentPrefix = rocketMQProperties.getEnvironmentPrefix(); + Assert.hasText(secretKey, "[spring.rocketmq.environmentPrefix] must not be null"); Properties producerProperties = new Properties(); - producerProperties.setProperty(PropertyKeyConst.ProducerId, "PID_"+groupName); + //生成者ProducerId添加前缀：PID_+环境标识_+groupName + producerProperties.setProperty(PropertyKeyConst.ProducerId, "PID_"+environmentPrefix+"_"+groupName); producerProperties.setProperty(PropertyKeyConst.AccessKey, accessKey); producerProperties.setProperty(PropertyKeyConst.SecretKey, secretKey); - producerProperties.setProperty(PropertyKeyConst.ONSAddr, rocketMQProperties.getNameServer()); - + producerProperties.setProperty(PropertyKeyConst.ONSAddr, onsAddr); Producer producer = ONSFactory.createProducer(producerProperties); return producer; } @@ -111,12 +112,13 @@ public class AliyunRocketMQAutoConfiguration { @Bean(destroyMethod = "destroy") @ConditionalOnBean(Producer.class) @ConditionalOnMissingBean(name = "rocketMQTemplate") - public RocketMQTemplate rocketMQTemplate(Producer mqProducer, + public RocketMQTemplate rocketMQTemplate(Producer mqProducer,RocketMQProperties rocketMQProperties, @Autowired(required = false) @Qualifier("rocketMQMessageObjectMapper") ObjectMapper objectMapper) { RocketMQTemplate rocketMQTemplate = new RocketMQTemplate(); rocketMQTemplate.setAliyunProducer(mqProducer); + rocketMQTemplate.setEnvironmentPrefix(rocketMQProperties.getEnvironmentPrefix()); if (Objects.nonNull(objectMapper)) { rocketMQTemplate.setObjectMapper(objectMapper); } @@ -176,10 +178,11 @@ public class AliyunRocketMQAutoConfiguration { RocketMQListener rocketMQListener = (RocketMQListener) bean; RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class); BeanDefinitionBuilder beanBuilder = BeanDefinitionBuilder.rootBeanDefinition(AliyunRocketMQListenerContainer.class); - beanBuilder.addPropertyValue(PROP_NAMESERVER, rocketMQProperties.getNameServer()); - beanBuilder.addPropertyValue(PROP_TOPIC, environment.resolvePlaceholders(annotation.topic())); + beanBuilder.addPropertyValue(PROP_ONS_Addr, rocketMQProperties.getOnsAddr()); + beanBuilder.addPropertyValue(PROP_TOPIC, rocketMQProperties.getEnvironmentPrefix()+"_"+environment.resolvePlaceholders(annotation.topic())); - beanBuilder.addPropertyValue(PROP_CONSUMER_GROUP, environment.resolvePlaceholders(annotation.consumerGroup())); + //消费者ConsumerId添加前缀：PID_+环境标识_+groupName + beanBuilder.addPropertyValue(PROP_CONSUMER_GROUP, "CID_"+rocketMQProperties.getEnvironmentPrefix()+"_"+environment.resolvePlaceholders(annotation.consumerGroup())); beanBuilder.addPropertyValue(PROP_CONSUME_MODE, annotation.consumeMode()); beanBuilder.addPropertyValue(PROP_CONSUME_THREAD_MAX, annotation.consumeThreadMax()); beanBuilder.addPropertyValue(PROP_MESSAGE_MODEL, annotation.messageModel()); @@ -187,13 +190,14 @@ public class AliyunRocketMQAutoConfiguration { beanBuilder.addPropertyValue(PROP_SELECTOR_TYPE, annotation.selectorType()); beanBuilder.addPropertyValue(PROP_ROCKETMQ_LISTENER, rocketMQListener); beanBuilder.addPropertyValue(PROP_ROCKETMQ_TEMPLATE, rocketMQTemplate); + beanBuilder.addPropertyValue(PROP_ENVIRONMENT_PREFIX, rocketMQProperties.getEnvironmentPrefix()); if (Objects.nonNull(objectMapper)) { beanBuilder.addPropertyValue(PROP_OBJECT_MAPPER, objectMapper); } beanBuilder.setDestroyMethodName(METHOD_DESTROY); //增加阿里云key - beanBuilder.addPropertyValue("accessKey", rocketMQProperties.getAccessKey()); - beanBuilder.addPropertyValue("secretKey", rocketMQProperties.getSecretKey()); + beanBuilder.addPropertyValue(PROP_ACCESS_KEY, rocketMQProperties.getAccessKey()); + beanBuilder.addPropertyValue(PROP_SECRET_KEY, rocketMQProperties.getSecretKey()); String containerBeanName = String.format("%s_%s", AliyunRocketMQListenerContainer.class.getName(), counter.incrementAndGet()); DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory(); diff --git a/src/main/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfiguration.java b/src/main/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfiguration.java deleted file mode 100644 index c945e96..0000000 --- a/src/main/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfiguration.java +++ /dev/null @@ -1,197 +0,0 @@ -/* - * 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.starter; - -import com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.rocketmq.spring.starter.annotation.RocketMQMessageListener; -import org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainer; -import org.apache.rocketmq.spring.starter.core.RocketMQListener; -import org.apache.rocketmq.spring.starter.core.RocketMQTemplate; -import java.util.Map; -import java.util.Objects; -import java.util.concurrent.atomic.AtomicLong; -import javax.annotation.Resource; -import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; -import org.apache.rocketmq.client.impl.MQClientAPIImpl; -import org.apache.rocketmq.client.producer.DefaultMQProducer; -import org.springframework.aop.support.AopUtils; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.DefaultListableBeanFactory; -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.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.core.annotation.Order; -import org.springframework.core.env.StandardEnvironment; -import org.springframework.util.Assert; - -import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.*; - -@Configuration -@ConditionalOnProperty(name = "spring.rocketmq.aliyun",havingValue="false") -@EnableConfigurationProperties(RocketMQProperties.class) -@ConditionalOnClass(MQClientAPIImpl.class) -@Order -@Slf4j -public class RocketMQAutoConfiguration { - - @Bean - @ConditionalOnClass(DefaultMQProducer.class) - @ConditionalOnMissingBean(DefaultMQProducer.class) - @ConditionalOnProperty(prefix = "spring.rocketmq", value = {"nameServer", "producer.group"}) - public DefaultMQProducer mqProducer(RocketMQProperties rocketMQProperties) { - - RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer(); - String groupName = producerConfig.getGroup(); - Assert.hasText(groupName, "[spring.rocketmq.producer.group] must not be null"); - - DefaultMQProducer producer = new DefaultMQProducer(producerConfig.getGroup()); - producer.setNamesrvAddr(rocketMQProperties.getNameServer()); - producer.setSendMsgTimeout(producerConfig.getSendMsgTimeout()); - producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed()); - producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed()); - producer.setMaxMessageSize(producerConfig.getMaxMessageSize()); - producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMsgBodyOverHowmuch()); - producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryAnotherBrokerWhenNotStoreOk()); - - return producer; - } - - @Bean - @ConditionalOnClass(ObjectMapper.class) - @ConditionalOnMissingBean(name = "rocketMQMessageObjectMapper") - public ObjectMapper rocketMQMessageObjectMapper() { - return new ObjectMapper(); - } - - @Bean(destroyMethod = "destroy") - @ConditionalOnBean(DefaultMQProducer.class) - @ConditionalOnMissingBean(name = "rocketMQTemplate") - public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer, - @Autowired(required = false) - @Qualifier("rocketMQMessageObjectMapper") - ObjectMapper objectMapper) { - RocketMQTemplate rocketMQTemplate = new RocketMQTemplate(); - rocketMQTemplate.setDefaultProducer(mqProducer); - if (Objects.nonNull(objectMapper)) { - rocketMQTemplate.setObjectMapper(objectMapper); - } - - return rocketMQTemplate; - } - - @Configuration - @ConditionalOnClass(DefaultMQPushConsumer.class) - @EnableConfigurationProperties(RocketMQProperties.class) - @ConditionalOnProperty(prefix = "spring.rocketmq", value = "nameServer") - @Order - public static class ListenerContainerConfiguration implements ApplicationContextAware, InitializingBean { - private ConfigurableApplicationContext applicationContext; - - private AtomicLong counter = new AtomicLong(0); - - @Resource - private StandardEnvironment environment; - - @Resource - private RocketMQProperties rocketMQProperties; - - private ObjectMapper objectMapper; - - @Autowired - private RocketMQTemplate rocketMQTemplate; - - public ListenerContainerConfiguration() { - } - - @Autowired(required = false) - public ListenerContainerConfiguration( - @Qualifier("rocketMQMessageObjectMapper") ObjectMapper objectMapper) { - this.objectMapper = objectMapper; - } - - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - this.applicationContext = (ConfigurableApplicationContext) applicationContext; - } - - @Override - public void afterPropertiesSet() { - Map beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class); - - if (Objects.nonNull(beans)) { - beans.forEach(this::registerContainer); - } - } - - private void registerContainer(String beanName, Object bean) { - Class clazz = AopUtils.getTargetClass(bean); - - if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) { - throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName()); - } - - RocketMQListener rocketMQListener = (RocketMQListener) bean; - RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class); - BeanDefinitionBuilder beanBuilder = BeanDefinitionBuilder.rootBeanDefinition(DefaultRocketMQListenerContainer.class); - beanBuilder.addPropertyValue(PROP_NAMESERVER, rocketMQProperties.getNameServer()); - beanBuilder.addPropertyValue(PROP_TOPIC, environment.resolvePlaceholders(annotation.topic())); - - beanBuilder.addPropertyValue(PROP_CONSUMER_GROUP, environment.resolvePlaceholders(annotation.consumerGroup())); - beanBuilder.addPropertyValue(PROP_CONSUME_MODE, annotation.consumeMode()); - beanBuilder.addPropertyValue(PROP_CONSUME_THREAD_MAX, annotation.consumeThreadMax()); - beanBuilder.addPropertyValue(PROP_MESSAGE_MODEL, annotation.messageModel()); - beanBuilder.addPropertyValue(PROP_SELECTOR_EXPRESS, environment.resolvePlaceholders(annotation.selectorExpress())); - beanBuilder.addPropertyValue(PROP_SELECTOR_TYPE, annotation.selectorType()); - beanBuilder.addPropertyValue(PROP_ROCKETMQ_LISTENER, rocketMQListener); - beanBuilder.addPropertyValue(PROP_ROCKETMQ_TEMPLATE, rocketMQTemplate); - if (Objects.nonNull(objectMapper)) { - beanBuilder.addPropertyValue(PROP_OBJECT_MAPPER, objectMapper); - } - beanBuilder.setDestroyMethodName(METHOD_DESTROY); - - String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(), counter.incrementAndGet()); - DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory(); - beanFactory.registerBeanDefinition(containerBeanName, beanBuilder.getBeanDefinition()); - - DefaultRocketMQListenerContainer container = beanFactory.getBean(containerBeanName, DefaultRocketMQListenerContainer.class); - - if (!container.isStarted()) { - try { - container.start(); - } catch (Exception e) { - log.error("started container failed. {}", container, e); - throw new RuntimeException(e); - } - } - - log.info("register rocketMQ listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName); - } - } -} diff --git a/src/main/java/org/apache/rocketmq/spring/starter/RocketMQProperties.java b/src/main/java/org/apache/rocketmq/spring/starter/RocketMQProperties.java index 7829ec4..c422df1 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/RocketMQProperties.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/RocketMQProperties.java @@ -24,22 +24,25 @@ import org.springframework.boot.context.properties.ConfigurationProperties; @ConfigurationProperties(prefix = "spring.rocketmq") @Data public class RocketMQProperties { - - /** - * name server for rocketMQ, formats: `host:port;host:port` + /** + * 环境前缀 + */ + private String environmentPrefix; + /** + * 消息队列服务接入点 */ - private String nameServer; + private String onsAddr; - private Producer producer; - /** - * 阿里云分配的accesskey + /** + * AccessKey, 用于标识、校验用户身份 */ private String accessKey; /** - * 阿里云分配的secretKey + * SecretKey, 用于标识、校验用户身份 */ private String secretKey; + private Producer producer; @Data public static class Producer { diff --git a/src/main/java/org/apache/rocketmq/spring/starter/annotation/RocketMQMessageListener.java b/src/main/java/org/apache/rocketmq/spring/starter/annotation/RocketMQMessageListener.java index 32ee587..c864dfa 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/annotation/RocketMQMessageListener.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/annotation/RocketMQMessageListener.java @@ -17,15 +17,17 @@ package org.apache.rocketmq.spring.starter.annotation; -import org.apache.rocketmq.common.filter.ExpressionType; -import org.apache.rocketmq.spring.starter.enums.ConsumeMode; -import org.apache.rocketmq.spring.starter.enums.SelectorType; import java.lang.annotation.Documented; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; + +import org.apache.rocketmq.spring.starter.enums.ConsumeMode; +import org.apache.rocketmq.spring.starter.enums.SelectorType; + +import com.aliyun.openservices.ons.api.ExpressionType; +import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel; @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) diff --git a/src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQListenerContainer.java b/src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQListenerContainer.java index 41664ef..52d3c72 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQListenerContainer.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQListenerContainer.java @@ -25,38 +25,39 @@ import java.util.List; import java.util.Objects; import java.util.Properties; -import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.exception.ExceptionUtils; -import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; -import org.apache.rocketmq.client.consumer.MessageSelector; -import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; -import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; -import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; -import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; -import org.apache.rocketmq.client.exception.MQClientException; -import org.apache.rocketmq.common.message.MessageExt; -import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; import org.apache.rocketmq.spring.starter.enums.ConsumeMode; import org.apache.rocketmq.spring.starter.enums.SelectorType; import org.apache.rocketmq.spring.starter.exception.ConvertMsgException; import org.apache.rocketmq.spring.starter.msgvo.ConsumeFailedMsgVO; +import org.apache.rocketmq.spring.starter.utils.ExceptionUtil; import org.apache.rocketmq.spring.starter.utils.IPUtil; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; import com.aliyun.openservices.ons.api.Action; import com.aliyun.openservices.ons.api.ConsumeContext; import com.aliyun.openservices.ons.api.Consumer; import com.aliyun.openservices.ons.api.Message; import com.aliyun.openservices.ons.api.MessageListener; +import com.aliyun.openservices.ons.api.MessageSelector; import com.aliyun.openservices.ons.api.ONSFactory; import com.aliyun.openservices.ons.api.PropertyKeyConst; import com.aliyun.openservices.ons.api.batch.BatchConsumer; -import com.aliyun.openservices.ons.api.bean.BatchConsumerBean; +import com.aliyun.openservices.ons.api.batch.BatchMessageListener; +import com.aliyun.openservices.ons.api.order.ConsumeOrderContext; import com.aliyun.openservices.ons.api.order.MessageOrderListener; +import com.aliyun.openservices.ons.api.order.OrderAction; import com.aliyun.openservices.ons.api.order.OrderConsumer; +import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer; +import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext; +import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.exception.MQClientException; +import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.MessageExt; +import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel; import com.fasterxml.jackson.databind.ObjectMapper; - +import static org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainerConstants.*; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; @@ -77,23 +78,13 @@ public class AliyunRocketMQListenerContainer implements InitializingBean, Rocket @Setter @Getter - private long suspendCurrentQueueTimeMillis = 1000; - + private String consumerGroup; /** - * Message consume retry strategy
-1,no retry,put into DLQ directly
0,broker control retry frequency
- * >0,client control retry frequency + * 消息队列服务接入点 */ @Setter @Getter - private int delayLevelWhenNextConsume = 0; - - @Setter - @Getter - private String consumerGroup; - - @Setter - @Getter - private String nameServer; + private String onsAddr; @Setter @Getter @@ -141,6 +132,11 @@ public class AliyunRocketMQListenerContainer implements InitializingBean, Rocket private BatchConsumer batchConsumer; private Class messageType; + /** + * 环境前缀 + */ + @Setter + private String environmentPrefix; @Setter private RocketMQTemplate rocketMQTemplate; @@ -180,66 +176,62 @@ public class AliyunRocketMQListenerContainer implements InitializingBean, Rocket @SuppressWarnings("unchecked") public Action consume(final Message message, final ConsumeContext context){ - for (MessageExt messageExt : msgs) { - Date consumeBeginTime = new Date(); - log.debug("received msg: {}", messageExt); - try { - long now = System.currentTimeMillis(); - rocketMQListener.onMessage(doConvertMessage(messageExt)); - long costTime = System.currentTimeMillis() - now; - log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime); - } catch (Exception e) { - log.warn("consume message failed. messageExt:{}", messageExt, e); - context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume); - if(messageExt.getTopic().equals("DATA_COLLECTION_TOPIC") && "ConsumeMsgFailed".equals(messageExt.getTags())){ - log.error("消费失败的消息为“保存消费失败日志消息”，不需要记录日志,不需要重新消费，直接返回成功"); - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; - } - if(e instanceof ConvertMsgException){ - log.error("消费失败的原因为转换对象失败，需要记录日志，不需要重新消费，返回消费成功"); - //消息消费失败，发送失败消息 - this.sendConsumeMsgFailed(messageExt,e,consumeBeginTime); - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; - } - this.sendConsumeMsgFailed(messageExt,e,consumeBeginTime); - return ConsumeConcurrentlyStatus.RECONSUME_LATER; + Date consumeBeginTime = new Date(); + log.debug("received msg: {}", message); + try { + long now = consumeBeginTime.getTime(); + rocketMQListener.onMessage(doConvertMessage(message)); + long costTime = System.currentTimeMillis() - now; + log.debug("consume {} cost: {} ms", message.getMsgID(), costTime); + } catch (Exception e) { + log.warn("consume message failed. message:{}", message, e); + if(message.getTopic().equals(environmentPrefix+"_"+CONSUMEFAILED_TOPIC) && CONSUMEFAILED_TAG.equals(message.getTag())){ + log.error("消费失败的消息为“保存消费失败日志消息”，不需要记录日志,不需要重新消费，直接返回成功"); + return Action.CommitMessage; + } + if(e instanceof ConvertMsgException){ + log.error("消费失败的原因为转换对象失败，需要记录日志，不需要重新消费，返回消费成功"); + //消息消费失败，发送失败消息 + this.sendConsumeMsgFailed(message,e,consumeBeginTime); + return Action.CommitMessage; } + this.sendConsumeMsgFailed(message,e,consumeBeginTime); + return Action.ReconsumeLater; } - - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + + return Action.CommitMessage; } /** * 发送消息消费失败消息 - * @param messageExt + * @param message * @param e * 2018年3月22日 zhaowg */ - private void sendConsumeMsgFailed(MessageExt messageExt, Exception e,Date consumeBeginTime) { + private void sendConsumeMsgFailed(Message message, Exception e,Date consumeBeginTime) { log.info("消费消息失败，开始发送消费失败MQ"); - String topic = "DATA_COLLECTION_TOPIC"; - String tag = "ConsumeMsgFailed"; + String topic = environmentPrefix+"_"+CONSUMEFAILED_TOPIC; + String tag = CONSUMEFAILED_TAG; try{ Date consumeEndTime = new Date(); - String destination = topic+":"+tag; ConsumeFailedMsgVO consumeFailedMsgVO = new ConsumeFailedMsgVO(); consumeFailedMsgVO.setConsumeBeginTime(consumeBeginTime); consumeFailedMsgVO.setConsumeEndTime(consumeEndTime); consumeFailedMsgVO.setConsumeGroup(consumerGroup); consumeFailedMsgVO.setConsumeIp(IPUtil.getLocalHost()); if(e!=null){ - String errMsg = ExceptionUtils.getStackTrace(e); - if(StringUtils.isNotBlank(errMsg)){ + String errMsg = ExceptionUtil.getTrace(e); + if(!StringUtils.isEmpty(errMsg)){ //最多保存1024个字符 consumeFailedMsgVO.setCunsumerErrMsg(errMsg.substring(0, 1024)); } } - consumeFailedMsgVO.setMsg(new String(messageExt.getBody())); - consumeFailedMsgVO.setMsgId(messageExt.getMsgId()); - consumeFailedMsgVO.setMsgKeys(messageExt.getKeys()); - consumeFailedMsgVO.setReconsumeTimes(messageExt.getReconsumeTimes()); - consumeFailedMsgVO.setTag(messageExt.getTags()); - consumeFailedMsgVO.setTopic(messageExt.getTopic()); - rocketMQTemplate.sendOneWay(destination, consumeFailedMsgVO); + consumeFailedMsgVO.setMsg(new String(message.getBody())); + consumeFailedMsgVO.setMsgId(message.getMsgID()); + consumeFailedMsgVO.setMsgKeys(message.getKey()); + consumeFailedMsgVO.setReconsumeTimes(message.getReconsumeTimes()); + consumeFailedMsgVO.setTag(message.getTag()); + consumeFailedMsgVO.setTopic(message.getTopic()); + rocketMQTemplate.sendOneWay(topic, tag, consumeFailedMsgVO); log.info("发送消息消费失败MQ成功"); }catch(Exception e1){ log.info("发送消息消费失败MQ异常",e); @@ -250,50 +242,103 @@ public class AliyunRocketMQListenerContainer implements InitializingBean, Rocket public class DefaultMessageListenerOrderly implements MessageOrderListener { - @SuppressWarnings("unchecked") - public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) { - for (MessageExt messageExt : msgs) { - log.debug("received msg: {}", messageExt); - try { - long now = System.currentTimeMillis(); - rocketMQListener.onMessage(doConvertMessage(messageExt)); - long costTime = System.currentTimeMillis() - now; - log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime); - } catch (Exception e) { - log.warn("consume message failed. messageExt:{}", messageExt, e); - context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis); - return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; - } + @Override + public OrderAction consume(Message message, ConsumeOrderContext context) { + log.debug("received msg: {}", message); + try { + long now = System.currentTimeMillis(); + rocketMQListener.onMessage(doConvertMessage(message)); + long costTime = System.currentTimeMillis() - now; + log.info("consume {} cost: {} ms", message.getMsgID(), costTime); + } catch (Exception e) { + log.warn("consume message failed. message:{}", message, e); + return OrderAction.Suspend; } - - return ConsumeOrderlyStatus.SUCCESS; - } + return OrderAction.Success; + } + } + + public class DefaultMessageListenerBatchs implements BatchMessageListener{ + + @Override + public Action consume(List messages, ConsumeContext context) { + for (Message message : messages) { + Date consumeBeginTime = new Date(); + log.debug("received msg: {}", message); + try { + long now = consumeBeginTime.getTime(); + rocketMQListener.onMessage(doConvertMessage(message)); + long costTime = System.currentTimeMillis() - now; + log.debug("consume {} cost: {} ms", message.getMsgID(), costTime); + } catch (Exception e) { + log.warn("consume message failed. message:{}", message, e); + if(message.getTopic().equals(environmentPrefix+"_"+CONSUMEFAILED_TOPIC) && CONSUMEFAILED_TAG.equals(message.getTag())){ + log.error("消费失败的消息为“保存消费失败日志消息”，不需要记录日志,不需要重新消费，直接返回成功"); + continue; + } + if(e instanceof ConvertMsgException){ + log.error("消费失败的原因为转换对象失败，需要记录日志，不需要重新消费，返回消费成功"); + //消息消费失败，发送失败消息 + this.sendConsumeMsgFailed(message,e,consumeBeginTime); + continue; + } + this.sendConsumeMsgFailed(message,e,consumeBeginTime); + return Action.ReconsumeLater; + } + } + return Action.CommitMessage; + } + + /** + * 发送消息消费失败消息 + * @param message + * @param e + * 2018年3月22日 zhaowg + */ + private void sendConsumeMsgFailed(Message message, Exception e,Date consumeBeginTime) { + log.info("消费消息失败，开始发送消费失败MQ"); + String topic = environmentPrefix+"_"+CONSUMEFAILED_TOPIC; + String tag = CONSUMEFAILED_TAG; + try{ + Date consumeEndTime = new Date(); + ConsumeFailedMsgVO consumeFailedMsgVO = new ConsumeFailedMsgVO(); + consumeFailedMsgVO.setConsumeBeginTime(consumeBeginTime); + consumeFailedMsgVO.setConsumeEndTime(consumeEndTime); + consumeFailedMsgVO.setConsumeGroup(consumerGroup); + consumeFailedMsgVO.setConsumeIp(IPUtil.getLocalHost()); + if(e!=null){ + String errMsg = ExceptionUtil.getTrace(e); + if(!StringUtils.isEmpty(errMsg)){ + //最多保存1024个字符 + consumeFailedMsgVO.setCunsumerErrMsg(errMsg.substring(0, 1024)); + } + } + consumeFailedMsgVO.setMsg(new String(message.getBody())); + consumeFailedMsgVO.setMsgId(message.getMsgID()); + consumeFailedMsgVO.setMsgKeys(message.getKey()); + consumeFailedMsgVO.setReconsumeTimes(message.getReconsumeTimes()); + consumeFailedMsgVO.setTag(message.getTag()); + consumeFailedMsgVO.setTopic(message.getTopic()); + rocketMQTemplate.sendOneWay(topic, tag, consumeFailedMsgVO); + log.info("发送消息消费失败MQ成功"); + }catch(Exception e1){ + log.info("发送消息消费失败MQ异常",e); + } + + } } - @Override public void afterPropertiesSet() throws Exception { start(); } - @Override - public String toString() { - return "DefaultRocketMQListenerContainer{" + - "consumerGroup='" + consumerGroup + '\'' + - ", nameServer='" + nameServer + '\'' + - ", topic='" + topic + '\'' + - ", consumeMode=" + consumeMode + - ", selectorType=" + selectorType + - ", selectorExpress='" + selectorExpress + '\'' + - ", messageModel=" + messageModel + - '}'; - } @SuppressWarnings("unchecked") - private Object doConvertMessage(MessageExt messageExt) { - if (Objects.equals(messageType, MessageExt.class)) { - return messageExt; + private Object doConvertMessage(Message message) { + if (Objects.equals(messageType, Message.class)) { + return message; } else { - String str = new String(messageExt.getBody(), Charset.forName(charset)); + String str = new String(message.getBody(), Charset.forName(charset)); if (Objects.equals(messageType, String.class)) { return str; } else { @@ -335,72 +380,45 @@ public class AliyunRocketMQListenerContainer implements InitializingBean, Rocket Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required"); Assert.notNull(consumerGroup, "Property 'consumerGroup' is required"); - Assert.notNull(nameServer, "Property 'nameServer' is required"); + Assert.notNull(onsAddr, "Property 'nameServer' is required"); Assert.notNull(topic, "Property 'topic' is required"); Properties consumerProperties = new Properties(); - consumerProperties.setProperty(PropertyKeyConst.ConsumerId, "CID_"+consumerGroup); + consumerProperties.setProperty(PropertyKeyConst.ConsumerId, consumerGroup); consumerProperties.setProperty(PropertyKeyConst.AccessKey, accessKey); consumerProperties.setProperty(PropertyKeyConst.SecretKey, secretKey); - consumerProperties.setProperty(PropertyKeyConst.ONSAddr, nameServer); + consumerProperties.setProperty(PropertyKeyConst.ONSAddr, onsAddr); consumerProperties.setProperty(PropertyKeyConst.ConsumeThreadNums, consumeThreadMax+""); consumerProperties.setProperty(PropertyKeyConst.MessageModel, messageModel.getModeCN()); - //判断是否为批量消费者 - boolean isBatchConsume = false; //允许用户自己设置该consumer的一些配置 - DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(); - if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) { - ((RocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(defaultMQPushConsumer); - isBatchConsume = defaultMQPushConsumer.getConsumeMessageBatchMaxSize()>1; + if (rocketMQListener instanceof AliyunRocketMQPushConsumerLifecycleListener) { + ((AliyunRocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumerProperties); } - switch (consumeMode) { case ORDERLY://顺序消息 orderConsumer = ONSFactory.createOrderedConsumer(consumerProperties); if(selectorType == SelectorType.TAG){ orderConsumer.subscribe(topic, selectorExpress, new DefaultMessageListenerOrderly()); }else if(selectorType == SelectorType.SQL92){ - orderConsumer.subscribe(topic, com.aliyun.openservices.ons.api.MessageSelector.bySql(selectorExpress), new DefaultMessageListenerOrderly()); + orderConsumer.subscribe(topic, MessageSelector.bySql(selectorExpress), new DefaultMessageListenerOrderly()); } break; case CONCURRENTLY://普通消息 - if(isBatchConsume){ - //批量消息 - - } - consumer = ONSFactory.createConsumer(consumerProperties); if(selectorType == SelectorType.TAG){ consumer.subscribe(topic, selectorExpress, new DefaultMessageListenerConcurrently()); }else if(selectorType == SelectorType.SQL92){ - consumer.subscribe(topic, com.aliyun.openservices.ons.api.MessageSelector.bySql(selectorExpress), new DefaultMessageListenerConcurrently()); + consumer.subscribe(topic, MessageSelector.bySql(selectorExpress), new DefaultMessageListenerConcurrently()); } break; + case BATCH://批量消息 + batchConsumer = ONSFactory.createBatchConsumer(consumerProperties); + batchConsumer.subscribe(topic, selectorExpress, new DefaultMessageListenerBatchs()); + break; default: throw new IllegalArgumentException("Property 'consumeMode' was wrong."); } - - - - - consumer.subscribe(MqConfig.TOPIC, MqConfig.TAG, new MessageListenerImpl()); - consumer.start(); - - switch (selectorType) { - case TAG: - consumer.subscribe(topic, selectorExpress); - break; - case SQL92: - consumer.subscribe(topic, MessageSelector.bySql(selectorExpress)); - break; - default: - throw new IllegalArgumentException("Property 'selectorType' was wrong."); - } - - - - } } diff --git a/src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQPushConsumerLifecycleListener.java b/src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQPushConsumerLifecycleListener.java new file mode 100644 index 0000000..e0b1860 --- /dev/null +++ b/src/main/java/org/apache/rocketmq/spring/starter/core/AliyunRocketMQPushConsumerLifecycleListener.java @@ -0,0 +1,23 @@ +/* + * 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.starter.core; + +import java.util.Properties; + +public interface AliyunRocketMQPushConsumerLifecycleListener extends RocketMQConsumerLifecycleListener { +} diff --git a/src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainer.java b/src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainer.java index cbed340..9e9889c 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainer.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainer.java @@ -1,355 +1,355 @@ -/* - * 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.starter.core; - -import java.lang.reflect.ParameterizedType; -import java.lang.reflect.Type; -import java.nio.charset.Charset; -import java.util.Date; -import java.util.List; -import java.util.Objects; - -import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.exception.ExceptionUtils; -import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; -import org.apache.rocketmq.client.consumer.MessageSelector; -import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; -import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; -import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; -import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; -import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; -import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; -import org.apache.rocketmq.client.exception.MQClientException; -import org.apache.rocketmq.common.message.MessageExt; -import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; -import org.apache.rocketmq.spring.starter.enums.ConsumeMode; -import org.apache.rocketmq.spring.starter.enums.SelectorType; -import org.apache.rocketmq.spring.starter.exception.ConvertMsgException; -import org.apache.rocketmq.spring.starter.msgvo.ConsumeFailedMsgVO; -import org.apache.rocketmq.spring.starter.utils.IPUtil; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.util.Assert; - -import com.fasterxml.jackson.databind.ObjectMapper; - -import lombok.Getter; -import lombok.Setter; -import lombok.extern.slf4j.Slf4j; - -@SuppressWarnings("WeakerAccess") -@Slf4j -public class DefaultRocketMQListenerContainer implements InitializingBean, RocketMQListenerContainer { - - @Setter - @Getter - private long suspendCurrentQueueTimeMillis = 1000; - - /** - * Message consume retry strategy
-1,no retry,put into DLQ directly
0,broker control retry frequency
- * >0,client control retry frequency - */ - @Setter - @Getter - private int delayLevelWhenNextConsume = 0; - - @Setter - @Getter - private String consumerGroup; - - @Setter - @Getter - private String nameServer; - - @Setter - @Getter - private String topic; - - @Setter - @Getter - private ConsumeMode consumeMode = ConsumeMode.CONCURRENTLY; - - @Setter - @Getter - private SelectorType selectorType = SelectorType.TAG; - - @Setter - @Getter - private String selectorExpress = "*"; - - @Setter - @Getter - private MessageModel messageModel = MessageModel.CLUSTERING; - - @Setter - @Getter - private int consumeThreadMax = 64; - - @Getter - @Setter - private String charset = "UTF-8"; - - @Setter - @Getter - private ObjectMapper objectMapper = new ObjectMapper(); - - @Setter - @Getter - private boolean started; - - @Setter - private RocketMQListener rocketMQListener; - - private DefaultMQPushConsumer consumer; - - private Class messageType; - - @Setter - private RocketMQTemplate rocketMQTemplate; - - public void setupMessageListener(RocketMQListener rocketMQListener) { - this.rocketMQListener = rocketMQListener; - } - - @Override - public void destroy() { - this.setStarted(false); - if (Objects.nonNull(consumer)) { - consumer.shutdown(); - } - log.info("container destroyed, {}", this.toString()); - } - - public synchronized void start() throws MQClientException { - - if (this.isStarted()) { - throw new IllegalStateException("container already started. " + this.toString()); - } - - initRocketMQPushConsumer(); - - // parse message type - this.messageType = getMessageType(); - log.debug("msgType: {}", messageType.getName()); - - consumer.start(); - this.setStarted(true); - - log.info("started container: {}", this.toString()); - } - - public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently { - - @SuppressWarnings("unchecked") - public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) { - for (MessageExt messageExt : msgs) { - Date consumeBeginTime = new Date(); - log.debug("received msg: {}", messageExt); - try { - long now = System.currentTimeMillis(); - rocketMQListener.onMessage(doConvertMessage(messageExt)); - long costTime = System.currentTimeMillis() - now; - log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime); - } catch (Exception e) { - log.warn("consume message failed. messageExt:{}", messageExt, e); - context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume); - if(messageExt.getTopic().equals("DATA_COLLECTION_TOPIC") && "ConsumeMsgFailed".equals(messageExt.getTags())){ - log.error("消费失败的消息为“保存消费失败日志消息”，不需要记录日志,不需要重新消费，直接返回成功"); - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; - } - if(e instanceof ConvertMsgException){ - log.error("消费失败的原因为转换对象失败，需要记录日志，不需要重新消费，返回消费成功"); - //消息消费失败，发送失败消息 - this.sendConsumeMsgFailed(messageExt,e,consumeBeginTime); - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; - } - this.sendConsumeMsgFailed(messageExt,e,consumeBeginTime); - return ConsumeConcurrentlyStatus.RECONSUME_LATER; - } - } - - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; - } - /** - * 发送消息消费失败消息 - * @param messageExt - * @param e - * 2018年3月22日 zhaowg - */ - private void sendConsumeMsgFailed(MessageExt messageExt, Exception e,Date consumeBeginTime) { - log.info("消费消息失败，开始发送消费失败MQ"); - String topic = "DATA_COLLECTION_TOPIC"; - String tag = "ConsumeMsgFailed"; - try{ - Date consumeEndTime = new Date(); - String destination = topic+":"+tag; - ConsumeFailedMsgVO consumeFailedMsgVO = new ConsumeFailedMsgVO(); - consumeFailedMsgVO.setConsumeBeginTime(consumeBeginTime); - consumeFailedMsgVO.setConsumeEndTime(consumeEndTime); - consumeFailedMsgVO.setConsumeGroup(consumerGroup); - consumeFailedMsgVO.setConsumeIp(IPUtil.getLocalHost()); - if(e!=null){ - String errMsg = ExceptionUtils.getStackTrace(e); - if(StringUtils.isNotBlank(errMsg)){ - //最多保存1024个字符 - consumeFailedMsgVO.setCunsumerErrMsg(errMsg.substring(0, 1024)); - } - } - consumeFailedMsgVO.setMsg(new String(messageExt.getBody())); - consumeFailedMsgVO.setMsgId(messageExt.getMsgId()); - consumeFailedMsgVO.setMsgKeys(messageExt.getKeys()); - consumeFailedMsgVO.setReconsumeTimes(messageExt.getReconsumeTimes()); - consumeFailedMsgVO.setTag(messageExt.getTags()); - consumeFailedMsgVO.setTopic(messageExt.getTopic()); - rocketMQTemplate.sendOneWay(destination, consumeFailedMsgVO); - log.info("发送消息消费失败MQ成功"); - }catch(Exception e1){ - log.info("发送消息消费失败MQ异常",e); - } - - } - } - - public class DefaultMessageListenerOrderly implements MessageListenerOrderly { - - @SuppressWarnings("unchecked") - public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) { - for (MessageExt messageExt : msgs) { - log.debug("received msg: {}", messageExt); - try { - long now = System.currentTimeMillis(); - rocketMQListener.onMessage(doConvertMessage(messageExt)); - long costTime = System.currentTimeMillis() - now; - log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime); - } catch (Exception e) { - log.warn("consume message failed. messageExt:{}", messageExt, e); - context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis); - return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; - } - } - - return ConsumeOrderlyStatus.SUCCESS; - } - } - - @Override - public void afterPropertiesSet() throws Exception { - start(); - } - - @Override - public String toString() { - return "DefaultRocketMQListenerContainer{" + - "consumerGroup='" + consumerGroup + '\'' + - ", nameServer='" + nameServer + '\'' + - ", topic='" + topic + '\'' + - ", consumeMode=" + consumeMode + - ", selectorType=" + selectorType + - ", selectorExpress='" + selectorExpress + '\'' + - ", messageModel=" + messageModel + - '}'; - } - - @SuppressWarnings("unchecked") - private Object doConvertMessage(MessageExt messageExt) { - if (Objects.equals(messageType, MessageExt.class)) { - return messageExt; - } else { - String str = new String(messageExt.getBody(), Charset.forName(charset)); - if (Objects.equals(messageType, String.class)) { - return str; - } else { - // if msgType not string, use objectMapper change it. - try { - return objectMapper.readValue(str, messageType); - } catch (Exception e) { - log.info("convert failed. str:{}, msgType:{}", str, messageType); - throw new ConvertMsgException("cannot convert message to " + messageType, e); - } - } - } - } - - private Class getMessageType() { - Type[] interfaces = rocketMQListener.getClass().getGenericInterfaces(); - if (Objects.nonNull(interfaces)) { - for (Type type : interfaces) { - if (type instanceof ParameterizedType) { - ParameterizedType parameterizedType = (ParameterizedType) type; - if (Objects.equals(parameterizedType.getRawType(), RocketMQListener.class)) { - Type[] actualTypeArguments = parameterizedType.getActualTypeArguments(); - if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) { - return (Class) actualTypeArguments[0]; - } else { - return Object.class; - } - } - } - } - - return Object.class; - } else { - return Object.class; - } - } - - private void initRocketMQPushConsumer() throws MQClientException { - - Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required"); - Assert.notNull(consumerGroup, "Property 'consumerGroup' is required"); - Assert.notNull(nameServer, "Property 'nameServer' is required"); - Assert.notNull(topic, "Property 'topic' is required"); - - consumer = new DefaultMQPushConsumer(consumerGroup); - consumer.setNamesrvAddr(nameServer); - consumer.setConsumeThreadMax(consumeThreadMax); - if (consumeThreadMax < consumer.getConsumeThreadMin()) { - consumer.setConsumeThreadMin(consumeThreadMax); - } - - consumer.setMessageModel(messageModel); - - switch (selectorType) { - case TAG: - consumer.subscribe(topic, selectorExpress); - break; - case SQL92: - consumer.subscribe(topic, MessageSelector.bySql(selectorExpress)); - break; - default: - throw new IllegalArgumentException("Property 'selectorType' was wrong."); - } - - switch (consumeMode) { - case ORDERLY: - consumer.setMessageListener(new DefaultMessageListenerOrderly()); - break; - case CONCURRENTLY: - consumer.setMessageListener(new DefaultMessageListenerConcurrently()); - break; - default: - throw new IllegalArgumentException("Property 'consumeMode' was wrong."); - } - - // provide an entryway to custom setting RocketMQ consumer - if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) { - ((RocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumer); - } - - } - -} +///* +// * 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.starter.core; +// +//import java.lang.reflect.ParameterizedType; +//import java.lang.reflect.Type; +//import java.nio.charset.Charset; +//import java.util.Date; +//import java.util.List; +//import java.util.Objects; +// +//import org.apache.commons.lang3.StringUtils; +//import org.apache.commons.lang3.exception.ExceptionUtils; +//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +//import org.apache.rocketmq.client.consumer.MessageSelector; +//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; +//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +//import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; +//import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +//import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; +//import org.apache.rocketmq.client.exception.MQClientException; +//import org.apache.rocketmq.common.message.MessageExt; +//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; +//import org.apache.rocketmq.spring.starter.enums.ConsumeMode; +//import org.apache.rocketmq.spring.starter.enums.SelectorType; +//import org.apache.rocketmq.spring.starter.exception.ConvertMsgException; +//import org.apache.rocketmq.spring.starter.msgvo.ConsumeFailedMsgVO; +//import org.apache.rocketmq.spring.starter.utils.IPUtil; +//import org.springframework.beans.factory.InitializingBean; +//import org.springframework.util.Assert; +// +//import com.fasterxml.jackson.databind.ObjectMapper; +// +//import lombok.Getter; +//import lombok.Setter; +//import lombok.extern.slf4j.Slf4j; +// +//@SuppressWarnings("WeakerAccess") +//@Slf4j +//public class DefaultRocketMQListenerContainer implements InitializingBean, RocketMQListenerContainer { +// +// @Setter +// @Getter +// private long suspendCurrentQueueTimeMillis = 1000; +// +// /** +// * Message consume retry strategy
-1,no retry,put into DLQ directly
0,broker control retry frequency
+// * >0,client control retry frequency +// */ +// @Setter +// @Getter +// private int delayLevelWhenNextConsume = 0; +// +// @Setter +// @Getter +// private String consumerGroup; +// +// @Setter +// @Getter +// private String nameServer; +// +// @Setter +// @Getter +// private String topic; +// +// @Setter +// @Getter +// private ConsumeMode consumeMode = ConsumeMode.CONCURRENTLY; +// +// @Setter +// @Getter +// private SelectorType selectorType = SelectorType.TAG; +// +// @Setter +// @Getter +// private String selectorExpress = "*"; +// +// @Setter +// @Getter +// private MessageModel messageModel = MessageModel.CLUSTERING; +// +// @Setter +// @Getter +// private int consumeThreadMax = 64; +// +// @Getter +// @Setter +// private String charset = "UTF-8"; +// +// @Setter +// @Getter +// private ObjectMapper objectMapper = new ObjectMapper(); +// +// @Setter +// @Getter +// private boolean started; +// +// @Setter +// private RocketMQListener rocketMQListener; +// +// private DefaultMQPushConsumer consumer; +// +// private Class messageType; +// +// @Setter +// private RocketMQTemplate rocketMQTemplate; +// +// public void setupMessageListener(RocketMQListener rocketMQListener) { +// this.rocketMQListener = rocketMQListener; +// } +// +// @Override +// public void destroy() { +// this.setStarted(false); +// if (Objects.nonNull(consumer)) { +// consumer.shutdown(); +// } +// log.info("container destroyed, {}", this.toString()); +// } +// +// public synchronized void start() throws MQClientException { +// +// if (this.isStarted()) { +// throw new IllegalStateException("container already started. " + this.toString()); +// } +// +// initRocketMQPushConsumer(); +// +// // parse message type +// this.messageType = getMessageType(); +// log.debug("msgType: {}", messageType.getName()); +// +// consumer.start(); +// this.setStarted(true); +// +// log.info("started container: {}", this.toString()); +// } +// +// public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently { +// +// @SuppressWarnings("unchecked") +// public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) { +// for (MessageExt messageExt : msgs) { +// Date consumeBeginTime = new Date(); +// log.debug("received msg: {}", messageExt); +// try { +// long now = System.currentTimeMillis(); +// rocketMQListener.onMessage(doConvertMessage(messageExt)); +// long costTime = System.currentTimeMillis() - now; +// log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime); +// } catch (Exception e) { +// log.warn("consume message failed. messageExt:{}", messageExt, e); +// context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume); +// if(messageExt.getTopic().equals("DATA_COLLECTION_TOPIC") && "ConsumeMsgFailed".equals(messageExt.getTags())){ +// log.error("消费失败的消息为“保存消费失败日志消息”，不需要记录日志,不需要重新消费，直接返回成功"); +// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; +// } +// if(e instanceof ConvertMsgException){ +// log.error("消费失败的原因为转换对象失败，需要记录日志，不需要重新消费，返回消费成功"); +// //消息消费失败，发送失败消息 +// this.sendConsumeMsgFailed(messageExt,e,consumeBeginTime); +// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; +// } +// this.sendConsumeMsgFailed(messageExt,e,consumeBeginTime); +// return ConsumeConcurrentlyStatus.RECONSUME_LATER; +// } +// } +// +// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; +// } +// /** +// * 发送消息消费失败消息 +// * @param messageExt +// * @param e +// * 2018年3月22日 zhaowg +// */ +// private void sendConsumeMsgFailed(MessageExt messageExt, Exception e,Date consumeBeginTime) { +// log.info("消费消息失败，开始发送消费失败MQ"); +// String topic = "DATA_COLLECTION_TOPIC"; +// String tag = "ConsumeMsgFailed"; +// try{ +// Date consumeEndTime = new Date(); +// String destination = topic+":"+tag; +// ConsumeFailedMsgVO consumeFailedMsgVO = new ConsumeFailedMsgVO(); +// consumeFailedMsgVO.setConsumeBeginTime(consumeBeginTime); +// consumeFailedMsgVO.setConsumeEndTime(consumeEndTime); +// consumeFailedMsgVO.setConsumeGroup(consumerGroup); +// consumeFailedMsgVO.setConsumeIp(IPUtil.getLocalHost()); +// if(e!=null){ +// String errMsg = ExceptionUtils.getStackTrace(e); +// if(StringUtils.isNotBlank(errMsg)){ +// //最多保存1024个字符 +// consumeFailedMsgVO.setCunsumerErrMsg(errMsg.substring(0, 1024)); +// } +// } +// consumeFailedMsgVO.setMsg(new String(messageExt.getBody())); +// consumeFailedMsgVO.setMsgId(messageExt.getMsgId()); +// consumeFailedMsgVO.setMsgKeys(messageExt.getKeys()); +// consumeFailedMsgVO.setReconsumeTimes(messageExt.getReconsumeTimes()); +// consumeFailedMsgVO.setTag(messageExt.getTags()); +// consumeFailedMsgVO.setTopic(messageExt.getTopic()); +// rocketMQTemplate.sendOneWay(destination, consumeFailedMsgVO); +// log.info("发送消息消费失败MQ成功"); +// }catch(Exception e1){ +// log.info("发送消息消费失败MQ异常",e); +// } +// +// } +// } +// +// public class DefaultMessageListenerOrderly implements MessageListenerOrderly { +// +// @SuppressWarnings("unchecked") +// public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) { +// for (MessageExt messageExt : msgs) { +// log.debug("received msg: {}", messageExt); +// try { +// long now = System.currentTimeMillis(); +// rocketMQListener.onMessage(doConvertMessage(messageExt)); +// long costTime = System.currentTimeMillis() - now; +// log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime); +// } catch (Exception e) { +// log.warn("consume message failed. messageExt:{}", messageExt, e); +// context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis); +// return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; +// } +// } +// +// return ConsumeOrderlyStatus.SUCCESS; +// } +// } +// +// @Override +// public void afterPropertiesSet() throws Exception { +// start(); +// } +// +// @Override +// public String toString() { +// return "DefaultRocketMQListenerContainer{" + +// "consumerGroup='" + consumerGroup + '\'' + +// ", nameServer='" + nameServer + '\'' + +// ", topic='" + topic + '\'' + +// ", consumeMode=" + consumeMode + +// ", selectorType=" + selectorType + +// ", selectorExpress='" + selectorExpress + '\'' + +// ", messageModel=" + messageModel + +// '}'; +// } +// +// @SuppressWarnings("unchecked") +// private Object doConvertMessage(MessageExt messageExt) { +// if (Objects.equals(messageType, MessageExt.class)) { +// return messageExt; +// } else { +// String str = new String(messageExt.getBody(), Charset.forName(charset)); +// if (Objects.equals(messageType, String.class)) { +// return str; +// } else { +// // if msgType not string, use objectMapper change it. +// try { +// return objectMapper.readValue(str, messageType); +// } catch (Exception e) { +// log.info("convert failed. str:{}, msgType:{}", str, messageType); +// throw new ConvertMsgException("cannot convert message to " + messageType, e); +// } +// } +// } +// } +// +// private Class getMessageType() { +// Type[] interfaces = rocketMQListener.getClass().getGenericInterfaces(); +// if (Objects.nonNull(interfaces)) { +// for (Type type : interfaces) { +// if (type instanceof ParameterizedType) { +// ParameterizedType parameterizedType = (ParameterizedType) type; +// if (Objects.equals(parameterizedType.getRawType(), RocketMQListener.class)) { +// Type[] actualTypeArguments = parameterizedType.getActualTypeArguments(); +// if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) { +// return (Class) actualTypeArguments[0]; +// } else { +// return Object.class; +// } +// } +// } +// } +// +// return Object.class; +// } else { +// return Object.class; +// } +// } +// +// private void initRocketMQPushConsumer() throws MQClientException { +// +// Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required"); +// Assert.notNull(consumerGroup, "Property 'consumerGroup' is required"); +// Assert.notNull(nameServer, "Property 'nameServer' is required"); +// Assert.notNull(topic, "Property 'topic' is required"); +// +// consumer = new DefaultMQPushConsumer(consumerGroup); +// consumer.setNamesrvAddr(nameServer); +// consumer.setConsumeThreadMax(consumeThreadMax); +// if (consumeThreadMax < consumer.getConsumeThreadMin()) { +// consumer.setConsumeThreadMin(consumeThreadMax); +// } +// +// consumer.setMessageModel(messageModel); +// +// switch (selectorType) { +// case TAG: +// consumer.subscribe(topic, selectorExpress); +// break; +// case SQL92: +// consumer.subscribe(topic, MessageSelector.bySql(selectorExpress)); +// break; +// default: +// throw new IllegalArgumentException("Property 'selectorType' was wrong."); +// } +// +// switch (consumeMode) { +// case ORDERLY: +// consumer.setMessageListener(new DefaultMessageListenerOrderly()); +// break; +// case CONCURRENTLY: +// consumer.setMessageListener(new DefaultMessageListenerConcurrently()); +// break; +// default: +// throw new IllegalArgumentException("Property 'consumeMode' was wrong."); +// } +// +// // provide an entryway to custom setting RocketMQ consumer +// if (rocketMQListener instanceof AliyunRocketMQPushConsumerLifecycleListener) { +// ((AliyunRocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumer); +// } +// +// } +// +//} diff --git a/src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainerConstants.java b/src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainerConstants.java index 861fb49..209630d 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainerConstants.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/core/DefaultRocketMQListenerContainerConstants.java @@ -32,6 +32,20 @@ public final class DefaultRocketMQListenerContainerConstants { public static final String PROP_ROCKETMQ_LISTENER = "rocketMQListener"; public static final String PROP_OBJECT_MAPPER = "objectMapper"; public static final String METHOD_DESTROY = "destroy"; - /**生产者 add zwg*/ - public static final String PROP_ROCKETMQ_TEMPLATE = "rocketMQTemplate"; + public static final String PROP_ROCKETMQ_TEMPLATE = "rocketMQTemplate"; + public static final String PROP_ONS_Addr = "onsAddr"; + public static final String PROP_ACCESS_KEY = "accessKey"; + public static final String PROP_SECRET_KEY = "secretKey"; + /** + * 环境前缀 + */ + public static final String PROP_ENVIRONMENT_PREFIX = "environmentPrefix"; + /** + * 消息消费失败发送的主题 + */ + public final static String CONSUMEFAILED_TOPIC = "ZTEITS_RNT_CLOUD"; + /** + * 消息消费失败发送的tag + */ + public final static String CONSUMEFAILED_TAG = "ConsumeMsgFailed"; } diff --git a/src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQPushConsumerLifecycleListener.java b/src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQPushConsumerLifecycleListener.java deleted file mode 100644 index e536947..0000000 --- a/src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQPushConsumerLifecycleListener.java +++ /dev/null @@ -1,23 +0,0 @@ -/* - * 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.starter.core; - -import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; - -public interface RocketMQPushConsumerLifecycleListener extends RocketMQConsumerLifecycleListener { -} diff --git a/src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQTemplate.java b/src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQTemplate.java index 9e2150d..5d88f62 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQTemplate.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/core/RocketMQTemplate.java @@ -17,54 +17,30 @@ package org.apache.rocketmq.spring.starter.core; -import com.aliyun.openservices.ons.api.MessageAccessor; -import com.aliyun.openservices.ons.api.OnExceptionContext; -import com.aliyun.openservices.ons.api.Producer; -import com.aliyun.openservices.ons.api.PropertyKeyConst; -import com.aliyun.openservices.ons.api.exception.ONSClientException; -import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.MessageExt; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; import java.nio.charset.Charset; -import java.util.Iterator; import java.util.Map; -import java.util.Objects; -import java.util.Properties; import java.util.Map.Entry; +import java.util.Objects; -import lombok.Getter; -import lombok.Setter; -import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.client.producer.DefaultMQProducer; -import org.apache.rocketmq.client.producer.MessageQueueSelector; -import org.apache.rocketmq.client.producer.SendCallback; -import org.apache.rocketmq.client.producer.SendResult; -import org.apache.rocketmq.client.producer.SendStatus; -import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash; -import org.apache.rocketmq.common.message.MessageConst; -import org.apache.rocketmq.common.message.MessageQueue; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.core.AbstractMessageSendingTemplate; -import org.springframework.messaging.core.MessagePostProcessor; -import org.springframework.messaging.support.MessageBuilder; -import org.springframework.util.Assert; -import org.springframework.util.MimeTypeUtils; -import org.springframework.util.StringUtils; -@SuppressWarnings({"WeakerAccess", "unused"}) +import com.aliyun.openservices.ons.api.Message; +import com.aliyun.openservices.ons.api.Producer; +import com.aliyun.openservices.ons.api.SendCallback; +import com.aliyun.openservices.ons.api.SendResult; +import com.fasterxml.jackson.databind.ObjectMapper; + +import lombok.Getter; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; + @Slf4j -public class RocketMQTemplate extends AbstractMessageSendingTemplate implements InitializingBean, DisposableBean { +public class RocketMQTemplate implements InitializingBean, DisposableBean { @Getter @Setter - private DefaultMQProducer defaultProducer; - - @Getter - @Setter private Producer aliyunProducer; @Setter @@ -74,479 +50,252 @@ public class RocketMQTemplate extends AbstractMessageSendingTemplate imp @Getter @Setter private String charset = "UTF-8"; - - @Getter + + /** + * 环境前缀 + */ @Setter - private MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash(); + private String environmentPrefix; /** - *

Send message in synchronous mode. This method returns only when the sending procedure totally completes. - * Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS - * notification, SMS marketing system, etc..

- * - * Warn: this method has internal retry-mechanism, that is, internal implementation will retry - * {@link DefaultMQProducer#getRetryTimesWhenSendFailed} times before claiming failure. As a result, multiple - * messages may potentially delivered to broker(s). It's up to the application developers to resolve potential - * duplication issue. - * - * @param destination formats: `topicName:tags` - * @param message {@link org.springframework.messaging.Message} + * 同步发送消息 + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param key 业务主键 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. + * @param userProperties 添加用户自定义属性键值对; 该键值对在消费消费时可被获取.也可用于做SQL属性过滤 + * @param startDeliverTime 设置消息的定时投递时间（绝对时间),最大延迟时间为7天. + *

+ *
    + *
  1. 延迟投递: 延迟3s投递, 设置为: System.currentTimeMillis() + 3000;
    2. + *
  2. 定时投递: 2016-02-01 11:30:00投递, 设置为: new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-02-01 11:30:00").getTime()
    3. + *
* @return {@link SendResult} + * 2018年3月23日 zhaowg */ - public SendResult syncSend(String destination, Message message) { - if (Objects.isNull(message) || Objects.isNull(message.getPayload())) { - log.info("syncSend failed. destination:{}, message is null ", destination); - throw new IllegalArgumentException("`message` and `message.payload` cannot be null"); + public SendResult syncSend(String topic,String tag,String keys,Object payload,Map userProperties,Long startDeliverTime) { + if (Objects.isNull(topic) || Objects.isNull(payload)) { + log.info("同步消息发送失败，主题和消息不能为空"); + throw new IllegalArgumentException("同步消息发送失败，主题和消息不能为空"); } try { - SendResult sendResult = new SendResult(); - long now = System.currentTimeMillis(); - if(aliyunProducer != null){ - //阿里云发送 - com.aliyun.openservices.ons.api.Message aliyunMsg = convertToAliyunRocketMsg(destination,message); - com.aliyun.openservices.ons.api.SendResult aliyunSendResult = aliyunProducer.send(aliyunMsg); - sendResult = convertAliyunSendResult(aliyunSendResult); - }else if(defaultProducer != null){ - org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message); - sendResult = defaultProducer.send(rocketMsg); - }else{ - throw new RuntimeException("product为空，请检查配置文件是否配置：spring.rocketmq.aliyun，且值为true或false"); - } + long now = System.currentTimeMillis(); + + Message rocketMsg = new Message(environmentPrefix+"_"+topic, tag, keys, convertToRocketMsg(payload)); + if(userProperties!=null && !userProperties.isEmpty()){ + for (Entry userProp : userProperties.entrySet()) { + rocketMsg.putUserProperties(userProp.getKey(), userProp.getValue()); + } + } + if(startDeliverTime!=null){ + //设置定时发送时间 + rocketMsg.setStartDeliverTime(startDeliverTime); + } + //阿里云发送 + SendResult sendResult = aliyunProducer.send(rocketMsg); long costTime = System.currentTimeMillis() - now; - log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId()); + log.debug("发送消息耗时： {} ms, msgId:{}", costTime, sendResult.getMessageId()); return sendResult; } catch (Exception e) { - log.info("syncSend failed. destination:{}, message:{} ", destination, message); + log.info("同步发送失败. topic:{}, message:{} ", topic, payload); throw new MessagingException(e.getMessage(), e); } } /** - * Same to {@link #syncSend(String, Message)}. - * - * @param destination formats: `topicName:tags` - * @param payload the Object to use as payload + * Same to {@link #syncSend(String, String, String, Object, Map, Long)}. + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param key 业务主键 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. * @return {@link SendResult} + * 2018年3月23日 zhaowg */ - public SendResult syncSend(String destination, Object payload) { - Message message = this.doConvert(payload, null, null); - return syncSend(destination, message); + public SendResult syncSend(String topic,String tag,String keys, Object payload) { + return syncSend(topic, tag, keys, payload, null, null); } - - - /** - * Same to {@link #syncSend(String, Message)} with send orderly with hashKey by specified. - * - * @param destination formats: `topicName:tags` - * @param message {@link org.springframework.messaging.Message} - * @param hashKey use this key to select queue. for example: orderId, productId ... - * @return {@link SendResult} - */ - /*public SendResult syncSendOrderly(String destination, Message message, String hashKey) { - if (Objects.isNull(message) || Objects.isNull(message.getPayload())) { - log.info("syncSendOrderly failed. destination:{}, message is null ", destination); - throw new IllegalArgumentException("`message` and `message.payload` cannot be null"); - } - - try { - long now = System.currentTimeMillis(); - org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message); - //TODO - throw new RuntimeException("暂时未整合阿里云Producer，不要使用"); -// SendResult sendResult = producer.send(rocketMsg, messageQueueSelector, hashKey, timeout); -// long costTime = System.currentTimeMillis() - now; -// log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId()); -// return sendResult; - } catch (Exception e) { - log.info("syncSendOrderly failed. destination:{}, message:{} ", destination, message); - throw new MessagingException(e.getMessage(), e); - } - }*/ - - /** - * Same to {@link #syncSend(String, Object)} with send orderly with hashKey by specified. - * - * @param destination formats: `topicName:tags` - * @param payload the Object to use as payload - * @param hashKey use this key to select queue. for example: orderId, productId ... + * Same to {@link #syncSend(String, String, String, Object)}. + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. * @return {@link SendResult} + * 2018年3月23日 zhaowg */ -// public SendResult syncSendOrderly(String destination, Object payload, String hashKey) { -// Message message = this.doConvert(payload, null, null); -// return syncSendOrderly(destination, message, hashKey); -// } + public SendResult syncSend(String topic,String tag, Object payload) { + return syncSend(topic, tag,null, payload); + } /** - * 将公共的sendCallBack转换为阿里云的sendCallBack - * @param sendCallback - * @return + * 异步发送消息 + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param key 业务主键 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. + * @param userProperties 添加用户自定义属性键值对; 该键值对在消费消费时可被获取.也可用于做SQL属性过滤 + * @param startDeliverTime 设置消息的定时投递时间（绝对时间),最大延迟时间为7天. + *

+ *
    + *
  1. 延迟投递: 延迟3s投递, 设置为: System.currentTimeMillis() + 3000;
    2. + *
  2. 定时投递: 2016-02-01 11:30:00投递, 设置为: new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-02-01 11:30:00").getTime()
    3. + *
+ * @param sendCallback 发送完成要执行的回调函数 * 2018年3月23日 zhaowg */ - private com.aliyun.openservices.ons.api.SendCallback aliyunSendCallBackConvert(final SendCallback sendCallback) { - com.aliyun.openservices.ons.api.SendCallback aliyunSendCallBack = new com.aliyun.openservices.ons.api.SendCallback() { - - @Override - public void onSuccess(com.aliyun.openservices.ons.api.SendResult sendResult) { - sendCallback.onSuccess(convertAliyunSendResult(sendResult)); - } - - @Override - public void onException(OnExceptionContext context) { - sendCallback.onException(context.getException()); - } - }; - return aliyunSendCallBack; - } - /** - *

Send message to broker asynchronously. asynchronous transmission is generally used in response time sensitive - * business scenarios.

- * - * This method returns immediately. On sending completion, sendCallback will be executed. - * - * Similar to {@link #syncSend(String, Object)}, internal implementation would potentially retry up to {@link - * DefaultMQProducer#getRetryTimesWhenSendAsyncFailed} times before claiming sending failure, which may yield - * message duplication and application developers are the one to resolve this potential issue. - * - * @param destination formats: `topicName:tags` - * @param message {@link org.springframework.messaging.Message} - * @param sendCallback {@link SendCallback} - */ - public void asyncSend(String destination, Message message, SendCallback sendCallback) { - if (Objects.isNull(message) || Objects.isNull(message.getPayload())) { - log.info("asyncSend failed. destination:{}, message is null ", destination); - throw new IllegalArgumentException("`message` and `message.payload` cannot be null"); + public void asyncSend(String topic,String tag,String keys,Object payload,Map userProperties, + Long startDeliverTime,SendCallback sendCallback) { + if (Objects.isNull(topic) || Objects.isNull(payload)) { + log.info("异步消息发送失败，主题和消息不能为空"); + throw new IllegalArgumentException("异步消息发送失败，主题和消息不能为空"); } - try { - if(aliyunProducer != null){ - com.aliyun.openservices.ons.api.Message aliyunMsg = this.convertToAliyunRocketMsg(destination, message); - aliyunProducer.sendAsync(aliyunMsg, aliyunSendCallBackConvert(sendCallback)); - }else if(defaultProducer != null){ - org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message); - defaultProducer.send(rocketMsg, sendCallback); - }else{ - throw new RuntimeException("product为空，请检查配置文件是否配置：spring.rocketmq.aliyun，且值为true或false"); + long now = System.currentTimeMillis(); + + Message rocketMsg = new Message(environmentPrefix+"_"+topic, tag, keys, convertToRocketMsg(payload)); + if(userProperties!=null && !userProperties.isEmpty()){ + for (Entry userProp : userProperties.entrySet()) { + rocketMsg.putUserProperties(userProp.getKey(), userProp.getValue()); + } + } + if(startDeliverTime!=null){ + //设置定时发送时间 + rocketMsg.setStartDeliverTime(startDeliverTime); } + //阿里云发送 + aliyunProducer.sendAsync(rocketMsg, sendCallback); + long costTime = System.currentTimeMillis() - now; + log.debug("发送消息耗时： {} ms", costTime); } catch (Exception e) { - log.info("asyncSend failed. destination:{}, message:{} ", destination, message); + log.info("异步发送失败. topic:{}, message:{} ", topic, payload); throw new MessagingException(e.getMessage(), e); } } - - /** - * Same to {@link #asyncSend(String, Message, SendCallback)}. - * - * @param destination formats: `topicName:tags` - * @param payload the Object to use as payload - * @param sendCallback {@link SendCallback} + /** + * Same to {@link #asyncSend(String, String, String, Object, Map, Long, SendCallback)}. + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param key 业务主键 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. + * @param sendCallback 发送完成要执行的回调函数 + * @return {@link SendResult} + * 2018年3月23日 zhaowg */ - public void asyncSend(String destination, Object payload, SendCallback sendCallback) { - Message message = this.doConvert(payload, null, null); - asyncSend(destination, message, sendCallback); + public void asyncSend(String topic,String tag,String keys, Object payload,SendCallback sendCallback) { + asyncSend(topic, tag, keys, payload, null, null,sendCallback); } - - /** - * Same to {@link #asyncSend(String, Message, SendCallback)} with send orderly with hashKey by specified. - * - * @param destination formats: `topicName:tags` - * @param message {@link org.springframework.messaging.Message} - * @param hashKey use this key to select queue. for example: orderId, productId ... - * @param sendCallback {@link SendCallback} - */ -// public void asyncSendOrderly(String destination, Message message, String hashKey, SendCallback sendCallback) { -// if (Objects.isNull(message) || Objects.isNull(message.getPayload())) { -// log.info("asyncSendOrderly failed. destination:{}, message is null ", destination); -// throw new IllegalArgumentException("`message` and `message.payload` cannot be null"); -// } -// -// try { -// org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message); -// //TODO zwg -// throw new RuntimeException("暂时未整合阿里云Producer，不要使用"); -// //producer.send(rocketMsg, messageQueueSelector, hashKey, sendCallback, timeout); -// } catch (Exception e) { -// log.info("asyncSendOrderly failed. destination:{}, message:{} ", destination, message); -// throw new MessagingException(e.getMessage(), e); -// } -// } - - /** - * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)}. - * - * @param destination formats: `topicName:tags` - * @param payload the Object to use as payload - * @param hashKey use this key to select queue. for example: orderId, productId ... - * @param sendCallback {@link SendCallback} + * Same to {@link #asyncSend(String, String, String, Object,SendCallback)}. + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. + * @param sendCallback 发送完成要执行的回调函数 + * @return {@link SendResult} + * 2018年3月23日 zhaowg */ -// public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback) { -// Message message = this.doConvert(payload, null, null); -// asyncSendOrderly(destination, message, hashKey, sendCallback); -// } - + public void asyncSend(String topic,String tag, Object payload,SendCallback sendCallback) { + asyncSend(topic, tag,null, payload,sendCallback); + } /** - * Similar to UDP, this method won't wait for - * acknowledgement from broker before return. Obviously, it has maximums throughput yet potentials of message loss. - * - * One-way transmission is used for cases requiring moderate reliability, such as log collection. - * - * @param destination formats: `topicName:tags` - * @param message {@link org.springframework.messaging.Message} + * 服务器不应答，无法保证消息是否成功到达服务器 + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param key 业务主键 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. + * @param userProperties 添加用户自定义属性键值对; 该键值对在消费消费时可被获取.也可用于做SQL属性过滤 + * @param startDeliverTime 设置消息的定时投递时间（绝对时间),最大延迟时间为7天. + *

+ *
    + *
  1. 延迟投递: 延迟3s投递, 设置为: System.currentTimeMillis() + 3000;
    2. + *
  2. 定时投递: 2016-02-01 11:30:00投递, 设置为: new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-02-01 11:30:00").getTime()
    3. + *
+ * 2018年3月23日 zhaowg */ - public void sendOneWay(String destination, Message message) { - if (Objects.isNull(message) || Objects.isNull(message.getPayload())) { - log.info("sendOneWay failed. destination:{}, message is null ", destination); - throw new IllegalArgumentException("`message` and `message.payload` cannot be null"); + public void sendOneWay(String topic,String tag,String keys,Object payload,Map userProperties, + Long startDeliverTime) { + if (Objects.isNull(topic) || Objects.isNull(payload)) { + log.info("sendOneWay消息发送失败，主题和消息不能为空"); + throw new IllegalArgumentException("sendOneWay消息发送失败，主题和消息不能为空"); } - try { - if(aliyunProducer !=null){ - //阿里云环境 - com.aliyun.openservices.ons.api.Message aliyunMsg = convertToAliyunRocketMsg(destination, message); - aliyunProducer.sendOneway(aliyunMsg); - }else if(defaultProducer != null){ - org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message); - defaultProducer.sendOneway(rocketMsg); - }else{ - throw new RuntimeException("product为空，请检查配置文件是否配置：spring.rocketmq.aliyun，且值为true或false"); - } + long now = System.currentTimeMillis(); + Message rocketMsg = new Message(environmentPrefix+"_"+topic, tag, keys, convertToRocketMsg(payload)); + if(userProperties!=null && !userProperties.isEmpty()){ + for (Entry userProp : userProperties.entrySet()) { + rocketMsg.putUserProperties(userProp.getKey(), userProp.getValue()); + } + } + if(startDeliverTime!=null){ + //设置定时发送时间 + rocketMsg.setStartDeliverTime(startDeliverTime); + } + //阿里云发送 + aliyunProducer.sendOneway(rocketMsg); + long costTime = System.currentTimeMillis() - now; + log.debug("发送消息耗时： {} ms", costTime); } catch (Exception e) { - log.info("sendOneWay failed. destination:{}, message:{} ", destination, message); + log.info("sendOneWay发送失败. topic:{}, message:{} ", topic, payload); throw new MessagingException(e.getMessage(), e); } } - - /** - * Same to {@link #sendOneWay(String, Message)} - * - * @param destination formats: `topicName:tags` - * @param payload the Object to use as payload + /** + * Same to {@link #sendOneWay(String, String, String, Object, Map, Long)}. + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param key 业务主键 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. + * 2018年3月23日 zhaowg */ - public void sendOneWay(String destination, Object payload) { - Message message = this.doConvert(payload, null, null); - sendOneWay(destination, message); + public void sendOneWay(String topic,String tag,String keys, Object payload) { + sendOneWay(topic, tag, keys, payload, null, null); } - /** - * Same to {@link #sendOneWay(String, Message)} with send orderly with hashKey by specified. - * - * @param destination formats: `topicName:tags` - * @param message {@link org.springframework.messaging.Message} - * @param hashKey use this key to select queue. for example: orderId, productId ... + * Same to {@link #sendOneWay(String, String, String, Object)}. + * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. + * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意 + * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. + * 2018年3月23日 zhaowg */ -// public void sendOneWayOrderly(String destination, Message message, String hashKey) { -// if (Objects.isNull(message) || Objects.isNull(message.getPayload())) { -// log.info("sendOneWayOrderly failed. destination:{}, message is null ", destination); -// throw new IllegalArgumentException("`message` and `message.payload` cannot be null"); -// } -// -// try { -// //TODO zwg -// throw new RuntimeException("暂时未整合阿里云Producer，不要使用"); -// //org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message); -// //producer.sendOneway(rocketMsg, messageQueueSelector, hashKey); -// } catch (Exception e) { -// log.info("sendOneWayOrderly failed. destination:{}, message:{}", destination, message); -// throw new MessagingException(e.getMessage(), e); -// } -// } + public void sendOneWay(String topic,String tag, Object payload) { + sendOneWay(topic, tag,null, payload); + } - /** - * Same to {@link #sendOneWayOrderly(String, Message, String)} - * - * @param destination formats: `topicName:tags` - * @param payload the Object to use as payload - */ -// public void sendOneWayOrderly(String destination, Object payload, String hashKey) { -// Message message = this.doConvert(payload, null, null); -// sendOneWayOrderly(destination, message, hashKey); -// } - @Override public void afterPropertiesSet() throws Exception { if(aliyunProducer != null){ - log.info("开始启动阿里云环境生产者"); + log.info("开始启动阿里云[环境标识："+environmentPrefix+"]生产者"); aliyunProducer.start(); - }else if(defaultProducer != null){ - log.info("开始启动非阿里云环境生产者"); - defaultProducer.start(); - }else{ - throw new RuntimeException("product为空，请检查配置文件是否配置：spring.rocketmq.aliyun，且值为true或false"); } } - protected void doSend(String destination, Message message) { - SendResult sendResult = syncSend(destination, message); - log.debug("send message to `{}` finished. result:{}", destination, sendResult); - } /** - * 转换阿里云返回对象 - * @param aliyunSendResult + * 转换对象为字节 + * @param msgObj * @return * 2018年3月23日 zhaowg */ - private SendResult convertAliyunSendResult(com.aliyun.openservices.ons.api.SendResult aliyunSendResult) { - SendResult sendResult = new SendResult(); - sendResult.setMsgId(aliyunSendResult.getMessageId()); - MessageQueue messageQueue = new MessageQueue(aliyunSendResult.getTopic(), null, 0); - sendResult.setMessageQueue(messageQueue); - sendResult.setSendStatus(SendStatus.SEND_OK); - return sendResult; - } - /** - * 转换为阿里云发送的消息对象 - * @param destination formats: `topicName:tags` - * @param message {@link org.springframework.messaging.Message} - * @return - * 2018年3月23日 zhaowg - */ - private com.aliyun.openservices.ons.api.Message convertToAliyunRocketMsg(String destination, Message message) { - Object payloadObj = message.getPayload(); - byte[] payloads; - - if (payloadObj instanceof String) { - payloads = ((String) payloadObj).getBytes(Charset.forName(charset)); - } else { - try { - String jsonObj = this.objectMapper.writeValueAsString(payloadObj); - payloads = jsonObj.getBytes(Charset.forName(charset)); - } catch (Exception e) { - throw new RuntimeException("convert to RocketMQ message failed.", e); - } - } - - String[] tempArr = destination.split(":", 2); - String topic = tempArr[0]; - String tags = ""; - if (tempArr.length > 1) { - tags = tempArr[1]; - } - - com.aliyun.openservices.ons.api.Message rocketMsg = new com.aliyun.openservices.ons.api.Message(topic, tags, payloads); - - MessageHeaders headers = message.getHeaders(); - if (Objects.nonNull(headers) && !headers.isEmpty()) { - Object keys = headers.get(MessageConst.PROPERTY_KEYS); - if (!StringUtils.isEmpty(keys)) { // if headers has 'KEYS', set rocketMQ message key - rocketMsg.setKey(keys.toString()); - } - - headers.entrySet().stream() - .filter(entry -> !Objects.equals(entry.getKey(), MessageConst.PROPERTY_KEYS) - && !Objects.equals(entry.getKey(), "FLAG") - && !Objects.equals(entry.getKey(), "WAIT_STORE_MSG_OK")) // exclude "KEYS", "FLAG", "WAIT_STORE_MSG_OK" - .forEach(entry -> { - rocketMsg.putUserProperties("USERS_" + entry.getKey(), String.valueOf(entry.getValue())); // add other properties with prefix "USERS_" - }); - - } - - return rocketMsg; - } - /** - * Convert spring message to rocketMQ message - * - * @param destination formats: `topicName:tags` - * @param message {@link org.springframework.messaging.Message} - * @return instance of {@link org.apache.rocketmq.common.message.Message} - */ - private org.apache.rocketmq.common.message.Message convertToRocketMsg(String destination, Message message) { - Object payloadObj = message.getPayload(); + private byte[] convertToRocketMsg(Object msgObj) { byte[] payloads; - if (payloadObj instanceof String) { - payloads = ((String) payloadObj).getBytes(Charset.forName(charset)); + if (msgObj instanceof String) { + payloads = ((String) msgObj).getBytes(Charset.forName(charset)); } else { try { - String jsonObj = this.objectMapper.writeValueAsString(payloadObj); + String jsonObj = this.objectMapper.writeValueAsString(msgObj); payloads = jsonObj.getBytes(Charset.forName(charset)); } catch (Exception e) { throw new RuntimeException("convert to RocketMQ message failed.", e); } } - - String[] tempArr = destination.split(":", 2); - String topic = tempArr[0]; - String tags = ""; - if (tempArr.length > 1) { - tags = tempArr[1]; - } - - org.apache.rocketmq.common.message.Message rocketMsg = new org.apache.rocketmq.common.message.Message(topic, tags, payloads); - - MessageHeaders headers = message.getHeaders(); - if (Objects.nonNull(headers) && !headers.isEmpty()) { - Object keys = headers.get(MessageConst.PROPERTY_KEYS); - if (!StringUtils.isEmpty(keys)) { // if headers has 'KEYS', set rocketMQ message key - rocketMsg.setKeys(keys.toString()); - } - - // set rocketMQ message flag - Object flagObj = headers.getOrDefault("FLAG", "0"); - int flag = 0; - try { - flag = Integer.parseInt(flagObj.toString()); - } catch (NumberFormatException e) { - // ignore - log.info("flag must be integer, flagObj:{}", flagObj); - } - rocketMsg.setFlag(flag); - - // set rocketMQ message waitStoreMsgOkObj - Object waitStoreMsgOkObj = headers.getOrDefault("WAIT_STORE_MSG_OK", "true"); - boolean waitStoreMsgOK = Boolean.TRUE.equals(waitStoreMsgOkObj); - rocketMsg.setWaitStoreMsgOK(waitStoreMsgOK); - - headers.entrySet().stream() - .filter(entry -> !Objects.equals(entry.getKey(), MessageConst.PROPERTY_KEYS) - && !Objects.equals(entry.getKey(), "FLAG") - && !Objects.equals(entry.getKey(), "WAIT_STORE_MSG_OK")) // exclude "KEYS", "FLAG", "WAIT_STORE_MSG_OK" - .forEach(entry -> { - rocketMsg.putUserProperty("USERS_" + entry.getKey(), String.valueOf(entry.getValue())); // add other properties with prefix "USERS_" - }); - - } - - return rocketMsg; + return payloads; } - @Override - protected Message doConvert(Object payload, Map headers, MessagePostProcessor postProcessor) { - String content; - if (payload instanceof String) { - content = (String) payload; - } else { - // if payload not as string, use objectMapper change it. - try { - content = objectMapper.writeValueAsString(payload); - } catch (JsonProcessingException e) { - log.info("convert payload to String failed. payload:{}", payload); - throw new RuntimeException("convert to payload to String failed.", e); - } - } - - MessageBuilder builder = MessageBuilder.withPayload(content); - if (headers != null) { - builder.copyHeaders(headers); - } - builder.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN); - - Message message = builder.build(); - if (postProcessor != null) { - message = postProcessor.postProcessMessage(message); - } - return message; - } @Override public void destroy() { - if (Objects.nonNull(defaultProducer)) { - defaultProducer.shutdown(); - } if(Objects.nonNull(aliyunProducer)){ + log.info("开始关闭阿里云[环境标识："+environmentPrefix+"]生产者"); aliyunProducer.shutdown(); } } diff --git a/src/main/java/org/apache/rocketmq/spring/starter/enums/ConsumeMode.java b/src/main/java/org/apache/rocketmq/spring/starter/enums/ConsumeMode.java index 569c861..04a0349 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/enums/ConsumeMode.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/enums/ConsumeMode.java @@ -19,12 +19,17 @@ package org.apache.rocketmq.spring.starter.enums; public enum ConsumeMode { /** - * receive asynchronously delivered messages concurrently + * 同时接收异步发送的消息 */ CONCURRENTLY, /** - * receive asynchronously delivered messages orderly. one queue, one thread + * 顺序接收消息，一个队列，一个线程 */ - ORDERLY + ORDERLY, + + /** + * 批量接收发送的消息，允许自定义范围为[1, 32], 实际消费数量可能小于该值 + */ + BATCH } diff --git a/src/main/java/org/apache/rocketmq/spring/starter/enums/SelectorType.java b/src/main/java/org/apache/rocketmq/spring/starter/enums/SelectorType.java index 561c64d..27604fe 100644 --- a/src/main/java/org/apache/rocketmq/spring/starter/enums/SelectorType.java +++ b/src/main/java/org/apache/rocketmq/spring/starter/enums/SelectorType.java @@ -17,7 +17,7 @@ package org.apache.rocketmq.spring.starter.enums; -import org.apache.rocketmq.common.filter.ExpressionType; +import com.aliyun.openservices.ons.api.ExpressionType; public enum SelectorType { diff --git a/src/main/java/org/apache/rocketmq/spring/starter/utils/ExceptionUtil.java b/src/main/java/org/apache/rocketmq/spring/starter/utils/ExceptionUtil.java new file mode 100644 index 0000000..e97da1c --- /dev/null +++ b/src/main/java/org/apache/rocketmq/spring/starter/utils/ExceptionUtil.java @@ -0,0 +1,21 @@ +package org.apache.rocketmq.spring.starter.utils; + +import java.io.PrintWriter; +import java.io.StringWriter; + +public class ExceptionUtil { + + public static String getTrace(Throwable t) { + StringBuffer buffer = new StringBuffer(); + if(t==null){ + return ""; + } + StringWriter stringWriter = new StringWriter(); + PrintWriter writer = new PrintWriter(stringWriter); + t.printStackTrace(writer); + //设置堆栈信息 + buffer.append("堆栈信息为:" + stringWriter.getBuffer().toString()); + return buffer.toString(); + } + +} diff --git a/src/test/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfigurationTests.java b/src/test/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfigurationTests.java index f599898..e4e2c41 100644 --- a/src/test/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfigurationTests.java +++ b/src/test/java/org/apache/rocketmq/spring/starter/RocketMQAutoConfigurationTests.java @@ -1,184 +1,184 @@ -/* - * 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.starter; - -import com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.rocketmq.spring.starter.annotation.RocketMQMessageListener; -import org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainer; -import org.apache.rocketmq.spring.starter.core.RocketMQListener; -import org.apache.rocketmq.spring.starter.core.RocketMQTemplate; -import org.apache.rocketmq.spring.starter.enums.ConsumeMode; -import org.apache.rocketmq.spring.starter.enums.SelectorType; -import org.apache.rocketmq.client.producer.DefaultMQProducer; -import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; -import org.junit.After; -import org.junit.Test; -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.boot.test.util.EnvironmentTestUtils; -import org.springframework.context.annotation.AnnotationConfigApplicationContext; - -import static org.assertj.core.api.Assertions.assertThat; - -public class RocketMQAutoConfigurationTests { - - private static final String TEST_CONSUMER_GROUP = "my_consumer"; - - private static final String TEST_TOPIC = "test-topic"; - - private AnnotationConfigApplicationContext context; - - @Test - public void rocketMQTemplate() { - - load("spring.rocketmq.nameServer=127.0.0.1:9876", - "spring.rocketmq.producer.group=my_group", - "spring.rocketmq.producer.send-msg-timeout=30000", - "spring.rocketmq.producer.retry-times-when-send-async-failed=1", - "spring.rocketmq.producer.compress-msg-body-over-howmuch=1024", - "spring.rocketmq.producer.max-message-size=10240", - "spring.rocketmq.producer.retry-another-broker-when-not-store-ok=true", - "spring.rocketmq.producer.retry-times-when-send-failed=1"); - - assertThat(this.context.containsBean("rocketMQMessageObjectMapper")).isTrue(); - assertThat(this.context.containsBean("mqProducer")).isTrue(); - assertThat(this.context.containsBean("rocketMQTemplate")).isTrue(); - assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); - - RocketMQTemplate rocketMQTemplate = this.context.getBean(RocketMQTemplate.class); - ObjectMapper objectMapper = this.context.getBean("rocketMQMessageObjectMapper", ObjectMapper.class); - assertThat(rocketMQTemplate.getObjectMapper()).isEqualTo(objectMapper); - - DefaultMQProducer defaultMQProducer = rocketMQTemplate.getProducer(); - - assertThat(defaultMQProducer.getNamesrvAddr()).isEqualTo("127.0.0.1:9876"); - assertThat(defaultMQProducer.getProducerGroup()).isEqualTo("my_group"); - assertThat(defaultMQProducer.getSendMsgTimeout()).isEqualTo(30000); - assertThat(defaultMQProducer.getRetryTimesWhenSendAsyncFailed()).isEqualTo(1); - assertThat(defaultMQProducer.getCompressMsgBodyOverHowmuch()).isEqualTo(1024); - assertThat(defaultMQProducer.getMaxMessageSize()).isEqualTo(10240); - assertThat(defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()).isTrue(); - assertThat(defaultMQProducer.getRetryTimesWhenSendFailed()).isEqualTo(1); - } - - @Test - public void enableProducer() { - load(); - assertThat(this.context.containsBean("mqProducer")).isFalse(); - assertThat(this.context.containsBean("rocketMQTemplate")).isFalse(); - closeContext(); - - load("spring.rocketmq.nameServer=127.0.0.1:9876"); - assertThat(this.context.containsBean("mqProducer")).isFalse(); - assertThat(this.context.containsBean("rocketMQTemplate")).isFalse(); - closeContext(); - - load("spring.rocketmq.producer.group=my_group"); - assertThat(this.context.containsBean("mqProducer")).isFalse(); - assertThat(this.context.containsBean("rocketMQTemplate")).isFalse(); - closeContext(); - - load("spring.rocketmq.nameServer=127.0.0.1:9876", "spring.rocketmq.producer.group=my_group"); - assertThat(this.context.containsBean("mqProducer")).isTrue(); - assertThat(this.context.containsBean("rocketMQTemplate")).isEqualTo(true); - assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); - } - - @Test - public void enableConsumer() { - load(); - assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); - closeContext(); - - load("spring.rocketmq.nameServer=127.0.0.1:9876"); - assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); - closeContext(); - - load(false); - this.context.registerBeanDefinition("myListener", - BeanDefinitionBuilder.rootBeanDefinition(MyListener.class).getBeanDefinition()); - this.context.refresh(); - assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); - closeContext(); - - load(false, "spring.rocketmq.nameServer=127.0.0.1:9876"); - this.context.registerBeanDefinition("myListener", - BeanDefinitionBuilder.rootBeanDefinition(MyListener.class).getBeanDefinition()); - this.context.refresh(); - assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isNotEmpty(); - assertThat(this.context.containsBean(DefaultRocketMQListenerContainer.class.getName() + "_1")).isTrue(); - assertThat(this.context.containsBean("mqProducer")).isFalse(); - assertThat(this.context.containsBean("rocketMQTemplate")).isFalse(); - - } - - @Test - public void listenerContainer() { - load(false, "spring.rocketmq.nameServer=127.0.0.1:9876"); - BeanDefinitionBuilder beanBuilder = BeanDefinitionBuilder.rootBeanDefinition(MyListener.class); - this.context.registerBeanDefinition("myListener", beanBuilder.getBeanDefinition()); - this.context.refresh(); - - assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isNotEmpty(); - assertThat(this.context.containsBean(DefaultRocketMQListenerContainer.class.getName() + "_1")).isTrue(); - - DefaultRocketMQListenerContainer listenerContainer = - this.context.getBean(DefaultRocketMQListenerContainer.class.getName() + "_1", - DefaultRocketMQListenerContainer.class); - ObjectMapper objectMapper = this.context.getBean("rocketMQMessageObjectMapper", ObjectMapper.class); - assertThat(listenerContainer.getObjectMapper()).isEqualTo(objectMapper); - assertThat(listenerContainer.getConsumeMode()).isEqualTo(ConsumeMode.CONCURRENTLY); - assertThat(listenerContainer.getSelectorType()).isEqualTo(SelectorType.TAG); - assertThat(listenerContainer.getSelectorExpress()).isEqualTo("*"); - assertThat(listenerContainer.getConsumerGroup()).isEqualTo(TEST_CONSUMER_GROUP); - assertThat(listenerContainer.getTopic()).isEqualTo(TEST_TOPIC); - assertThat(listenerContainer.getNameServer()).isEqualTo("127.0.0.1:9876"); - assertThat(listenerContainer.getMessageModel()).isEqualTo(MessageModel.CLUSTERING); - assertThat(listenerContainer.getConsumeThreadMax()).isEqualTo(1); - } - - @After - public void closeContext() { - if (this.context != null) { - this.context.close(); - } - } - - @RocketMQMessageListener(consumerGroup = TEST_CONSUMER_GROUP, topic = TEST_TOPIC, consumeThreadMax = 1) - private static class MyListener implements RocketMQListener { - - @Override - public void onMessage(String message) { - System.out.println(message); - } - } - - private void load(boolean refresh, String... environment) { - AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(); - ctx.register(RocketMQAutoConfiguration.class); - EnvironmentTestUtils.addEnvironment(ctx, environment); - if (refresh) { - ctx.refresh(); - } - this.context = ctx; - } - - private void load(String... environment) { - load(true, environment); - } -} - +///* +// * 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.starter; +// +//import com.fasterxml.jackson.databind.ObjectMapper; +//import org.apache.rocketmq.spring.starter.annotation.RocketMQMessageListener; +//import org.apache.rocketmq.spring.starter.core.DefaultRocketMQListenerContainer; +//import org.apache.rocketmq.spring.starter.core.RocketMQListener; +//import org.apache.rocketmq.spring.starter.core.RocketMQTemplate; +//import org.apache.rocketmq.spring.starter.enums.ConsumeMode; +//import org.apache.rocketmq.spring.starter.enums.SelectorType; +//import org.apache.rocketmq.client.producer.DefaultMQProducer; +//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; +//import org.junit.After; +//import org.junit.Test; +//import org.springframework.beans.factory.support.BeanDefinitionBuilder; +//import org.springframework.boot.test.util.EnvironmentTestUtils; +//import org.springframework.context.annotation.AnnotationConfigApplicationContext; +// +//import static org.assertj.core.api.Assertions.assertThat; +// +//public class RocketMQAutoConfigurationTests { +// +// private static final String TEST_CONSUMER_GROUP = "my_consumer"; +// +// private static final String TEST_TOPIC = "test-topic"; +// +// private AnnotationConfigApplicationContext context; +// +// @Test +// public void rocketMQTemplate() { +// +// load("spring.rocketmq.nameServer=127.0.0.1:9876", +// "spring.rocketmq.producer.group=my_group", +// "spring.rocketmq.producer.send-msg-timeout=30000", +// "spring.rocketmq.producer.retry-times-when-send-async-failed=1", +// "spring.rocketmq.producer.compress-msg-body-over-howmuch=1024", +// "spring.rocketmq.producer.max-message-size=10240", +// "spring.rocketmq.producer.retry-another-broker-when-not-store-ok=true", +// "spring.rocketmq.producer.retry-times-when-send-failed=1"); +// +// assertThat(this.context.containsBean("rocketMQMessageObjectMapper")).isTrue(); +// assertThat(this.context.containsBean("mqProducer")).isTrue(); +// assertThat(this.context.containsBean("rocketMQTemplate")).isTrue(); +// assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); +// +// RocketMQTemplate rocketMQTemplate = this.context.getBean(RocketMQTemplate.class); +// ObjectMapper objectMapper = this.context.getBean("rocketMQMessageObjectMapper", ObjectMapper.class); +// assertThat(rocketMQTemplate.getObjectMapper()).isEqualTo(objectMapper); +// +// DefaultMQProducer defaultMQProducer = rocketMQTemplate.getProducer(); +// +// assertThat(defaultMQProducer.getNamesrvAddr()).isEqualTo("127.0.0.1:9876"); +// assertThat(defaultMQProducer.getProducerGroup()).isEqualTo("my_group"); +// assertThat(defaultMQProducer.getSendMsgTimeout()).isEqualTo(30000); +// assertThat(defaultMQProducer.getRetryTimesWhenSendAsyncFailed()).isEqualTo(1); +// assertThat(defaultMQProducer.getCompressMsgBodyOverHowmuch()).isEqualTo(1024); +// assertThat(defaultMQProducer.getMaxMessageSize()).isEqualTo(10240); +// assertThat(defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()).isTrue(); +// assertThat(defaultMQProducer.getRetryTimesWhenSendFailed()).isEqualTo(1); +// } +// +// @Test +// public void enableProducer() { +// load(); +// assertThat(this.context.containsBean("mqProducer")).isFalse(); +// assertThat(this.context.containsBean("rocketMQTemplate")).isFalse(); +// closeContext(); +// +// load("spring.rocketmq.nameServer=127.0.0.1:9876"); +// assertThat(this.context.containsBean("mqProducer")).isFalse(); +// assertThat(this.context.containsBean("rocketMQTemplate")).isFalse(); +// closeContext(); +// +// load("spring.rocketmq.producer.group=my_group"); +// assertThat(this.context.containsBean("mqProducer")).isFalse(); +// assertThat(this.context.containsBean("rocketMQTemplate")).isFalse(); +// closeContext(); +// +// load("spring.rocketmq.nameServer=127.0.0.1:9876", "spring.rocketmq.producer.group=my_group"); +// assertThat(this.context.containsBean("mqProducer")).isTrue(); +// assertThat(this.context.containsBean("rocketMQTemplate")).isEqualTo(true); +// assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); +// } +// +// @Test +// public void enableConsumer() { +// load(); +// assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); +// closeContext(); +// +// load("spring.rocketmq.nameServer=127.0.0.1:9876"); +// assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); +// closeContext(); +// +// load(false); +// this.context.registerBeanDefinition("myListener", +// BeanDefinitionBuilder.rootBeanDefinition(MyListener.class).getBeanDefinition()); +// this.context.refresh(); +// assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isEmpty(); +// closeContext(); +// +// load(false, "spring.rocketmq.nameServer=127.0.0.1:9876"); +// this.context.registerBeanDefinition("myListener", +// BeanDefinitionBuilder.rootBeanDefinition(MyListener.class).getBeanDefinition()); +// this.context.refresh(); +// assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isNotEmpty(); +// assertThat(this.context.containsBean(DefaultRocketMQListenerContainer.class.getName() + "_1")).isTrue(); +// assertThat(this.context.containsBean("mqProducer")).isFalse(); +// assertThat(this.context.containsBean("rocketMQTemplate")).isFalse(); +// +// } +// +// @Test +// public void listenerContainer() { +// load(false, "spring.rocketmq.nameServer=127.0.0.1:9876"); +// BeanDefinitionBuilder beanBuilder = BeanDefinitionBuilder.rootBeanDefinition(MyListener.class); +// this.context.registerBeanDefinition("myListener", beanBuilder.getBeanDefinition()); +// this.context.refresh(); +// +// assertThat(this.context.getBeansOfType(DefaultRocketMQListenerContainer.class)).isNotEmpty(); +// assertThat(this.context.containsBean(DefaultRocketMQListenerContainer.class.getName() + "_1")).isTrue(); +// +// DefaultRocketMQListenerContainer listenerContainer = +// this.context.getBean(DefaultRocketMQListenerContainer.class.getName() + "_1", +// DefaultRocketMQListenerContainer.class); +// ObjectMapper objectMapper = this.context.getBean("rocketMQMessageObjectMapper", ObjectMapper.class); +// assertThat(listenerContainer.getObjectMapper()).isEqualTo(objectMapper); +// assertThat(listenerContainer.getConsumeMode()).isEqualTo(ConsumeMode.CONCURRENTLY); +// assertThat(listenerContainer.getSelectorType()).isEqualTo(SelectorType.TAG); +// assertThat(listenerContainer.getSelectorExpress()).isEqualTo("*"); +// assertThat(listenerContainer.getConsumerGroup()).isEqualTo(TEST_CONSUMER_GROUP); +// assertThat(listenerContainer.getTopic()).isEqualTo(TEST_TOPIC); +// assertThat(listenerContainer.getNameServer()).isEqualTo("127.0.0.1:9876"); +// assertThat(listenerContainer.getMessageModel()).isEqualTo(MessageModel.CLUSTERING); +// assertThat(listenerContainer.getConsumeThreadMax()).isEqualTo(1); +// } +// +// @After +// public void closeContext() { +// if (this.context != null) { +// this.context.close(); +// } +// } +// +// @RocketMQMessageListener(consumerGroup = TEST_CONSUMER_GROUP, topic = TEST_TOPIC, consumeThreadMax = 1) +// private static class MyListener implements RocketMQListener { +// +// @Override +// public void onMessage(String message) { +// System.out.println(message); +// } +// } +// +// private void load(boolean refresh, String... environment) { +// AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(); +// ctx.register(RocketMQAutoConfiguration.class); +// EnvironmentTestUtils.addEnvironment(ctx, environment); +// if (refresh) { +// ctx.refresh(); +// } +// this.context = ctx; +// } +// +// private void load(String... environment) { +// load(true, environment); +// } +//} +// -- libgit2 0.21.4