【移動(dòng)應(yīng)用開(kāi)發(fā)技術(shù)】SpringBoot整合RocketMQ遇到的坑怎么解決

【移動(dòng)應(yīng)用開(kāi)發(fā)技術(shù)】SpringBoot整合RocketMQ遇到的坑怎么解決

本篇內(nèi)容主要講解“SpringBoot整合RocketMQ遇到的坑怎么解決”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓在下來(lái)帶大家學(xué)習(xí)“SpringBoot整合RocketMQ遇到的坑怎么解決”吧!在實(shí)現(xiàn)RocketMQ消費(fèi)時(shí)，一般會(huì)用到@RocketMQMessageListener注解定義Group、Topic以及selectorExpression（數(shù)據(jù)過(guò)濾、選擇的規(guī)則）為了能支持動(dòng)態(tài)篩選數(shù)據(jù)，一般都會(huì)使用表達(dá)式，然后通過(guò)apollo或者cloudconfig進(jìn)行動(dòng)態(tài)切換。

<!--

RocketMq

Spring

Boot

Starter-->

<dependency>

<groupId>org.apache.rocketmq</groupId>

<artifactId>rocketmq-spring-boot-starter</artifactId>

<version>2.0.4</version>

</dependency>@RocketMQMessageListener(consumerGroup

=

"${rocketmq.group}",topic

="${rocketmq.topic}",selectorExpression

=

"${rocketmq.selectorExpression}")

public

class

Consumer

implements

RocketMQListener<String>

{

@Override

public

void

onMessage(String

s)

{

System.out.println("消費(fèi)到的數(shù)據(jù)為："+s);

}

}RocketMQMessageListener整個(gè)注解默認(rèn)selectorExpression為*，表示接收當(dāng)前Topic下的所有數(shù)據(jù)，如果我們想對(duì)tags進(jìn)行動(dòng)態(tài)配置，在使用${rocketmq.selectorExpression}表達(dá)式時(shí)會(huì)發(fā)現(xiàn)所有數(shù)據(jù)全被過(guò)濾了，跟蹤源碼（ListenerContainerConfiguration.java）發(fā)現(xiàn)在創(chuàng)建listener時(shí)selectorExpression的數(shù)據(jù)在通environment環(huán)境變量中獲取對(duì)應(yīng)的數(shù)據(jù)后又被覆蓋了，導(dǎo)致整個(gè)過(guò)濾條件被變更為表達(dá)式。@Override

public

void

afterSingletonsInstantiated()

{

//

獲取所有所有使用了RocketMQMessageListener注解的bean

Map<String,

Object>

beans

=

this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class);

if

(Objects.nonNull(beans))

{

//

循環(huán)注冊(cè)容器

beans.forEach(this::registerContainer);

}

}

private

void

registerContainer(String

beanName,

Object

bean)

{

Class<?>

clazz

=

AopProxyUtils.ultimateTargetClass(bean);

//

校驗(yàn)當(dāng)前bean是否實(shí)現(xiàn)了RocketMQListener接口

if

(!RocketMQListener.class.isAssignableFrom(bean.getClass()))

{

throw

new

IllegalStateException(clazz

+

"

is

not

instance

of

"

+

RocketMQListener.class.getName());

}

//

獲取bean上的annotation

RocketMQMessageListener

annotation

=

clazz.getAnnotation(RocketMQMessageListener.class);

//

解析group及topic，可支持表達(dá)式

String

consumerGroup

=

this.environment.resolvePlaceholders(annotation.consumerGroup());

String

topic

=

this.environment.resolvePlaceholders(annotation.topic());

boolean

listenerEnabled

=

(boolean)rocketMQProperties.getConsumer().getListeners().getOrDefault(consumerGroup,

Collections.EMPTY_MAP)

.getOrDefault(topic,

true);

if

(!listenerEnabled)

{

log.debug(

"Consumer

Listener

(group:{},topic:{})

is

not

enabled

by

configuration,

will

ignore

initialization.",

consumerGroup,

topic);

return;

}

validate(annotation);

String

containerBeanName

=

String.format("%s_%s",

DefaultRocketMQListenerContainer.class.getName(),

counter.incrementAndGet());

GenericApplicationContext

genericApplicationContext

=

(GenericApplicationContext)applicationContext;

//

注冊(cè)bean的，調(diào)用createRocketMQListenerContainer

genericApplicationContext.registerBean(containerBeanName,

DefaultRocketMQListenerContainer.class,

()

->

createRocketMQListenerContainer(containerBeanName,

bean,

annotation));

DefaultRocketMQListenerContainer

container

=

genericApplicationContext.getBean(containerBeanName,

DefaultRocketMQListenerContainer.class);

if

(!container.isRunning())

{

try

{

container.start();

}

catch

(Exception

e)

{

log.error("Started

container

failed.

{}",

container,

e);

throw

new

RuntimeException(e);

}

}

("Register

the

listener

to

container,

listenerBeanName:{},

containerBeanName:{}",

beanName,

containerBeanName);

}

private

DefaultRocketMQListenerContainer

createRocketMQListenerContainer(String

name,

Object

bean,

RocketMQMessageListener

annotation)

{

DefaultRocketMQListenerContainer

container

=

new

DefaultRocketMQListenerContainer();

container.setRocketMQMessageListener(annotation);

String

nameServer

=

environment.resolvePlaceholders(Server());

nameServer

=

StringUtils.isEmpty(nameServer)

?

rocketMQProperties.getNameServer()

:

nameServer;

String

accessChannel

=

environment.resolvePlaceholders(annotation.accessChannel());

container.setNameServer(nameServer);

if

(!StringUtils.isEmpty(accessChannel))

{

container.setAccessChannel(AccessChannel.valueOf(accessChannel));

}

container.setTopic(environment.resolvePlaceholders(annotation.topic()));

//

此處已經(jīng)根據(jù)表達(dá)式將數(shù)據(jù)取出

String

tags

=

environment.resolvePlaceholders(annotation.selectorExpression());

if

(!StringUtils.isEmpty(tags))

{

container.setSelectorExpression(tags);

}

container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup()));

//

此處將SelectorExpression的數(shù)據(jù)覆蓋成了表達(dá)式

container.setRocketMQMessageListener(annotation);

container.setRocketMQListener((RocketMQListener)bean);

container.setObjectMapper(objectMapper);

container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());

container.setName(name);

//

REVIEW

ME,

use

the

same

clientId

or

multiple?

return

container;

}因?yàn)長(zhǎng)istenerContainerConfiguration類(lèi)是實(shí)現(xiàn)了SmartInitializingSingleton接口的afterSingletonsInstantiated方法，我們可以通過(guò)反射對(duì)selectorExpression的數(shù)據(jù)在ListenerContainerConfiguration進(jìn)行初始化前進(jìn)行解析并賦值回去。/**

*

在springboot初始化后，RocketMQ容器初始化前利用反射動(dòng)態(tài)改變數(shù)據(jù)

**/

@Configuration

public

class

ChangeSelectorExpressionBeforeMQInit

implements

InitializingBean

{

@Autowired

private

ApplicationContext

applicationContext;

@Autowired

private

StandardEnvironment

environment;

@Override

public

void

afterPropertiesSet()

throws

Exception

{

Map<String,Object>

beans

=applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class);

for

(Object

bean

:

beans.values()){

Class<?>

clazz

=

AopProxyUtils.ultimateTargetClass(bean);

if

(!RocketMQListener.class.isAssignableFrom(bean.getClass()))

{

continue;

}

RocketMQMessageListener

annotation

=

clazz.getAnnotation(RocketMQMess