Spring延时初始化RocketMQ消费者

小鱼儿 2021-11-27 03:36 574阅读 0赞

`Spring`加载`RocketMq`消费者实例后会立即开始消费，不论`Spring`容器是否初始化完毕。这就意味着如果`Spring`容器由于某种原因导致容器初始化失败，但应用程序没有因为容器初始化失败而关闭，那么会出现不可预期的后果，例如`Spring`容器初始化失败，导致定时任务不能正常执行，而`MQ`消费需要定时任务从数据库中拉取的配置信息，这样会导致`MQ`消费失败。因此，可以在`Spring`容器启动完成后再初始化`MQ`消费者。

通过`XML`或`Java`配置`Bean`，都可以指定`lazy-init`的属性，它有`true`和`false`两个值

true：延时加载，当有其他Bean有引用此Bean时或者主动调用BeanFactory#getBean方法时才会进行初始化
    false：立即加载

如果你只使用了`lazy-init = true`，`MQ`消费者并不会启动，如上所说，**当有其他Bean有引用此Bean时或者主动调用BeanFactory\#getBean方法才会进行初始化**。

**延时初始化MQ消费者示例：**

配置

<bean name="messageListenerImpl" class="com.MessageListenerImpl"/>
    
    <bean id="consumer" class="org.apache.rocketmq.client.consumer.DefaultMQPushConsumer" init-method="start"
    	  destroy-method="shutdown" lazy-init="true">
    	<property name="namesrvAddr" value="1.1.1.1:9876"/>
    	<property name="instanceName" value="push"/>
    	<property name="consumerGroup" value="CONSUMER_GROUP"/>
    	<property name="subscription">
    		<map>
    			<entry key="TOPIC" value="TAGS"/>
    		</map>
    	</property>
    	<property name="messageListener" ref="messageListenerImpl"/>
    	<property name="consumeThreadMin" value="2"/>
    	<property name="consumeThreadMax" value="4"/>
    	<property name="consumeMessageBatchMaxSize" value="1"/>
    </bean>

消费监听

public class MessageListenerImpl implements MessageListenerConcurrently {
    
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            try {
                for (MessageExt message : msgs) {
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }

使用`Spring`容器监听器，在容器启动成功后初始化`MQ`消费者

// 一个应用中可以有多个ApplicationListener，先加载的先执行
    @Component
    public class RocketMqInit implements ApplicationListener<ContextStartedEvent> {
        @Resource
        private ApplicationContext applicationContext;
        @Override
        public void onApplicationEvent(ContextStartedEvent event) {
            applicationContext.getBean(DefaultMQPushConsumer.class);
        }
    }