Springboot Rabbitmq + 线程池技术控制指定数量task执行

定义DataSyncTaskManager，作为线程池任务控制器

package org.demo.scheduletest.service;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class DataSyncTaskManager {
    // 线程数
    private static final Integer threadNum = 5;

    private static DataSyncTaskManager taskManager = null;
    private static BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();

    private ThreadPoolExecutor taskExecutorPool;

    private DataSyncTaskManager() {
        taskExecutorPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(threadNum);
    }

    /**
     * 构建唯一Manager对象单例
     *
     * @return
     */
    public static synchronized DataSyncTaskManager getManager() {
        if (null == taskManager) {
            taskManager = new DataSyncTaskManager();
        }

        return taskManager;
    }

    /**
     * 提交需要运行的任务
     *
     * @param task
     */
    public void submitTask(DataSyncTask task) {
        taskQueue.add(task);
        log.info("[DataSyncTaskManager] submitTask size={}", taskQueue.size());
    }

    public void runTaskDaemon() {
        log.info("[DataSyncTaskManager] runTaskDaemon start.");
        Thread thread = new Thread(() -> {
            while (true) {
                try {
                    Runnable task = taskQueue.take();
                    taskExecutorPool.submit(task);
                    // log.info("[DataSyncTaskManager] runTaskDaemon submit task={}", task);

                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("[startTaskRunningDaemon] task run InterruptedException", e);
                } catch (Exception e) {
                    log.error("[startTaskRunningDaemon] task run Exception", e);
                }
            }
        });

        thread.setName(this.getClass().getSimpleName());
        thread.start();
    }
}

定义DataSyncTask，作为具体任务执行方

package org.demo.scheduletest.service;

import com.rabbitmq.client.Channel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;


@Data
@Slf4j
public class DataSyncTask implements Runnable {
    private String name;
    private Channel channel;
    private long deliveryTag;

    public DataSyncTask(String name, Channel channel, long deliveryTag) {
        this.name = name;
        this.channel = channel;
        this.deliveryTag = deliveryTag;
    }

    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        log.info("[DataSyncTask] run task start, name = {}", name);

        try {
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        try {
            channel.basicAck(deliveryTag, true);
            log.info("[DataSyncTask] run task end, name = {}", name);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

}

InitTask，服务启动执行Task管理器

package org.demo.scheduletest.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

/**
 * @author zhe.xiao
 * @version 1.0
 * @since 2025/1/9 上午11:38
 */
@Slf4j
@Component
public class InitTask implements ApplicationRunner {
    /**
     * Callback used to run the bean.
     *
     * @param args incoming application arguments
     * @throws Exception on error
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        DataSyncTaskManager.getManager().runTaskDaemon();
    }
}

配置Rabbitmq

package org.demo.scheduletest.rabbitmq;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;

/**
 * @author zhe.xiao
 * @date 2022-07-06 17:27
 * @description
 **/
@SpringBootConfiguration
public class MyRabbitTemplateConfig {
    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host:/}")
    private String virtualhost;

    /**
     * 连接工厂
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualhost);
        //connectionFactory.setPublisherConfirms(true);
        return connectionFactory;
    }

    /**
     *
     * @return RabbitTemplate
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        return new RabbitTemplate(connectionFactory());
    }
}

package org.demo.scheduletest.rabbitmq;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;

/**
 * @author zhe.xiao
 * @date 2022-07-05 14:55
 * @description
 **/
@Configuration
public class MyRabbitExecutor {
    //正常队列
    public static final String QUEUE_1 = "my:queue:1";
    public static final String EXCHANGE_1 = "my:exchange:1";
    public static final String ROUTEING_1 = "data:route:1";

    //死信队列
    public static final String QUEUE_DEAD_LETTER = "my:queue:deadLetter";
    public static final String EXCHANGE_DEAD_LETTER = "my:exchange:deadLetter";
    public static final String ROUTING_DEAD_LETTER = "data:route:deadLetter";

    // 提供 Queue
    @Bean
    Queue myQueue1(){
        HashMap<String, Object> args = new HashMap<>();
        //绑定死信队列信息
        args.put("x-dead-letter-exchange", EXCHANGE_DEAD_LETTER);
        args.put("x-dead-letter-routing-key", ROUTING_DEAD_LETTER);


//        args.put("x-max-length", 5); //队列最大长度,超过了会进入死信队列
//         args.put("x-message-ttl", 5000); //如果5秒没被消费，则进入死信队列

        return new Queue(QUEUE_1, true, false, false, args);
    }

    // 提供 Exchange
    @Bean
    DirectExchange myExchange1(){
        return new DirectExchange(EXCHANGE_1, true, false);
    }

    // 创建一个Binding对象，将Exchange和Queue绑定在一起
    @Bean
    Binding myBinding1(){
        return BindingBuilder.bind(myQueue1()).to(myExchange1()).with(ROUTEING_1);
        // return BindingBuilder.bind(myQueue1()).to(myExchange1());
    }

    // 死信队列配置 QUEUE, EXCHANGE, BINDING
    @Bean
    Queue myQueueDeadLetter(){
        return new Queue(QUEUE_DEAD_LETTER, true, false, false);
    }

    @Bean
    DirectExchange myExchangeDeadLetter(){
        return new DirectExchange(EXCHANGE_DEAD_LETTER, true, false);
    }

    @Bean
    Binding myBindingDeadLetter(){
        return BindingBuilder.bind(myQueueDeadLetter()).to(myExchangeDeadLetter()).with(ROUTING_DEAD_LETTER);
    }
}

Rabbitmq消费者通过task控制器提交执行任务

package org.demo.scheduletest.rabbitmq;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.demo.scheduletest.service.DataSyncTask;
import org.demo.scheduletest.service.DataSyncTaskManager;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 * 消费QUEUE
 *
 * @author zhe.xiao
 * @date 2022-07-05 14:57
 * @description
 **/
@Slf4j
@Component
public class MyReceiver {
    @RabbitListener(queues = MyRabbitExecutor.QUEUE_1)
    public void handler1(String data, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.info("handler1 process: {}", data);
            DataSyncTask dataSyncTask = new DataSyncTask(data, channel, deliveryTag);
            DataSyncTaskManager.getManager().submitTask(dataSyncTask);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }
}