推荐订阅源

一、引入场景启动器

 <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.2.3</version>

        </dependency>

或者

 <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.3.0</version>

        </dependency>

 2、添加相关配置

rocketmq:
  name-server: ${ROCKETMQ_NAME_SERVER:192.168.0.24:30094}
  producer:
    group: ${ROCKETMQ_PRODUCER_GROUP:product_group}
    send-message-timeout: ${ROCKETMQ_SEND_MESSAGE_TIMEOUT:30000}
    max-message-size: ${ROCKETMQ_MAX_MESSAGE_SIZE:8388608}

 3、添加监听

package jnpf.qualifications.consummer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;


@Component
@RocketMQMessageListener(topic = "permission-topic", consumerGroup = "consumer_group", replyTimeout = 600000)
@Slf4j
public class TestMQListener implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {


    @Override
    public void onMessage(String message) {
        log.error("原始数据入库消费消息为空，过期或者被清除，任务Id为:{}", message);
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        // 重试次数
        consumer.setMaxReconsumeTimes(10);
  
    }

}