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()*/@Overridepublic 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*/@Overridepublic 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*/@Beanpublic ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);connectionFactory.setUsername(username);connectionFactory.setPassword(password);connectionFactory.setVirtualHost(virtualhost);//connectionFactory.setPublisherConfirms(true);return connectionFactory;}/**** @return RabbitTemplate*/@Beanpublic 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@BeanQueue 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@BeanDirectExchange myExchange1(){return new DirectExchange(EXCHANGE_1, true, false);}// 创建一个Binding对象，将Exchange和Queue绑定在一起@BeanBinding myBinding1(){return BindingBuilder.bind(myQueue1()).to(myExchange1()).with(ROUTEING_1);// return BindingBuilder.bind(myQueue1()).to(myExchange1());}// 死信队列配置 QUEUE, EXCHANGE, BINDING@BeanQueue myQueueDeadLetter(){return new Queue(QUEUE_DEAD_LETTER, true, false, false);}@BeanDirectExchange myExchangeDeadLetter(){return new DirectExchange(EXCHANGE_DEAD_LETTER, true, false);}@BeanBinding 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());}}
}