推荐订阅源

首先定义一个消费listener：

package demo;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class RabbitmqConsumer {

    @RabbitListener(queues = "visit_api", concurrency="1", autoStartup = "false", id="rabbit_listener_test")
    public void processMsg(String msg) {
        //TODO: 消息持久化
    }

}

然后实现一个启动类，实现ApplicationRunner接口：

package demo;

import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

/**
 * Rabbit消息队列启动类。
 */
@Component
public class RabbitApplicationRunner implements ApplicationRunner {

    private final RabbitListenerEndpointRegistry registry;

    @Autowired
    public RabbitApplicationRunner(RabbitListenerEndpointRegistry registry) {
        this.registry =  registry;
    }

    @Override
    public void run(ApplicationArguments args) {
        String[] listeners = new String[] { "rabbit_listener_test" };
        for (String listenerId: listeners) {
            MessageListenerContainer container = registry.getListenerContainer(listenerId);
            if (container == null) {
                System.out.println(listenerId + " is null.");
            } else {
                if (container.isRunning()) {
                    System.out.println(listenerId +" is running.");
                } else {
                    container.start();
                    System.out.println(listenerId + " started.");
                }
            }
        }
    }

}