Rocketmq整合Spring中的推消费和litepull消费

StringNAME_SERVER_PLACEHOLDER="${rocketmq.name-server:}";
StringACCESS_KEY_PLACEHOLDER="${rocketmq.push-consumer.access-key:}";
StringSECRET_KEY_PLACEHOLDER="${rocketmq.push-consumer.secret-key:}";
StringTRACE_TOPIC_PLACEHOLDER="${rocketmq.push-consumer.customized-trace-topic:}";
StringACCESS_CHANNEL_PLACEHOLDER="${rocketmq.access-channel:}";
StringconsumerGroup();
Stringtopic();
SelectorTypeselectorType() defaultSelectorType.TAG;
StringselectorExpression() default"*";
ConsumeModeconsumeMode() defaultConsumeMode.CONCURRENTLY;
MessageModelmessageModel() defaultMessageModel.CLUSTERING;
@DeprecatedintconsumeThreadMax() default64;
intconsumeThreadNumber() default20;
intmaxReconsumeTimes() default-1;
longconsumeTimeout() default15L;
intreplyTimeout() default3000;
StringaccessKey() defaultACCESS_KEY_PLACEHOLDER;
StringsecretKey() defaultSECRET_KEY_PLACEHOLDER;
booleanenableMsgTrace() defaultfalse;
StringcustomizedTraceTopic() defaultTRACE_TOPIC_PLACEHOLDER;
StringnameServer() defaultNAME_SERVER_PLACEHOLDER;
StringaccessChannel() defaultACCESS_CHANNEL_PLACEHOLDER;
StringtlsEnable() default"false";
Stringnamespace() default"";
/*** Message consume retry strategy in concurrently mode.** -1,no retry,put into DLQ directly* 0,broker control retry frequency* >0,client control retry frequency*/intdelayLevelWhenNextConsume() default0;
intsuspendCurrentQueueTimeMillis() default1000;
intawaitTerminationMillisWhenShutdown() default1000;
StringinstanceName() default"DEFAULT";

@OverridepublicvoidafterSingletonsInstantiated() {
//拿到带RocketMQMessageListener的注解，同时将其进行注册Map<String, Object>beans=this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class)
            .entrySet().stream().filter(entry->!ScopedProxyUtils.isScopedTarget(entry.getKey()))
            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
beans.forEach(this::registerContainer);
    }

privatevoidregisterContainer(StringbeanName, Objectbean) {
Class<?>clazz=AopProxyUtils.ultimateTargetClass(bean);
//拿到监听，获取消费者组、主题RocketMQMessageListenerannotation=clazz.getAnnotation(RocketMQMessageListener.class);
StringconsumerGroup=this.environment.resolvePlaceholders(annotation.consumerGroup());
Stringtopic=this.environment.resolvePlaceholders(annotation.topic());
//获取监听，是否开启了监听booleanlistenerEnabled=            (boolean) rocketMQProperties.getPushConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
                .getOrDefault(topic, true);
//容器bean名称StringcontainerBeanName=String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
counter.incrementAndGet());
GenericApplicationContextgenericApplicationContext= (GenericApplicationContext) applicationContext;
genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
            () ->createRocketMQListenerContainer(containerBeanName, bean, annotation));
DefaultRocketMQListenerContainercontainer=genericApplicationContext.getBean(containerBeanName,
DefaultRocketMQListenerContainer.class);
if (!container.isRunning()) {
try {
//启动容器container.start();
            } catch (Exceptione) {
log.error("Started container failed. {}", container, e);
thrownewRuntimeException(e);
            }
        }
log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
    }

@Overridepublicvoidstart() throwsMQClientException {
setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
//启动默认推模式的消费方式this.defaultMQPushConsumerImpl.start();
if (null!=traceDispatcher) {
try {
//链路追踪traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
            } catch (MQClientExceptione) {
log.warn("trace dispatcher start failed ", e);
            }
        }
    }

@Bean(CONSUMER_BEAN_NAME)
@ConditionalOnMissingBean(DefaultLitePullConsumer.class)
@ConditionalOnProperty(prefix="rocketmq", value= {"name-server", "pull-consumer.group", "pull-consumer.topic"})
publicDefaultLitePullConsumerdefaultLitePullConsumer(RocketMQPropertiesrocketMQProperties)
throwsMQClientException {
RocketMQProperties.PullConsumerconsumerConfig=rocketMQProperties.getPullConsumer();
StringnameServer=rocketMQProperties.getNameServer();
StringgroupName=consumerConfig.getGroup();
StringtopicName=consumerConfig.getTopic();
Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
Assert.hasText(groupName, "[rocketmq.pull-consumer.group] must not be null");
Assert.hasText(topicName, "[rocketmq.pull-consumer.topic] must not be null");
StringaccessChannel=rocketMQProperties.getAccessChannel();
MessageModelmessageModel=MessageModel.valueOf(consumerConfig.getMessageModel());
SelectorTypeselectorType=SelectorType.valueOf(consumerConfig.getSelectorType());
StringselectorExpression=consumerConfig.getSelectorExpression();
Stringak=consumerConfig.getAccessKey();
Stringsk=consumerConfig.getSecretKey();
intpullBatchSize=consumerConfig.getPullBatchSize();
booleanuseTLS=consumerConfig.isTlsEnable();
DefaultLitePullConsumerlitePullConsumer=RocketMQUtil.createDefaultLitePullConsumer(nameServer, accessChannel,
groupName, topicName, messageModel, selectorType, selectorExpression, ak, sk, pullBatchSize, useTLS);
litePullConsumer.setEnableMsgTrace(consumerConfig.isEnableMsgTrace());
litePullConsumer.setCustomizedTraceTopic(consumerConfig.getCustomizedTraceTopic());
litePullConsumer.setNamespace(consumerConfig.getNamespace());
litePullConsumer.setInstanceName(consumerConfig.getInstanceName());
log.info(String.format("a pull consumer(%s sub %s) init on namesrv %s",  groupName, topicName,nameServer));
returnlitePullConsumer;
    }

@ExtRocketMQConsumerConfiguration(topic="${demo.rocketmq.topic}", group="string_consumer", tlsEnable="${demo.ext.consumer.tlsEnable}")
publicclassExtRocketMQTemplateextendsRocketMQTemplate {
}

privatevoidregisterTemplate(StringbeanName, Objectbean) {
Class<?>clazz=AopProxyUtils.ultimateTargetClass(bean);
// 获取扩展配置消费者的配置ExtRocketMQConsumerConfigurationannotation=clazz.getAnnotation(ExtRocketMQConsumerConfiguration.class);
GenericApplicationContextgenericApplicationContext= (GenericApplicationContext) applicationContext;
DefaultLitePullConsumerconsumer=null;
try {
//创建消费者consumer=createConsumer(annotation);
//启动消费者DefaultLitePullConsumerconsumer.start();
        } catch (Exceptione) {
log.error("Failed to startup PullConsumer for RocketMQTemplate {}", beanName, e);
        }
RocketMQTemplaterocketMQTemplate= (RocketMQTemplate) bean;
rocketMQTemplate.setConsumer(consumer);
rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
log.info("Set real consumer to :{} {}", beanName, annotation.value());
    }

privateDefaultMQProducerproducer;
privateDefaultLitePullConsumerconsumer;