当前位置：首页 > news >正文

线程池中任务队列满-如何把多余的任务存储到Redis中

news 来源：原创 2025/5/20 10:00:24

当线程池的任务队列满时，可以采取以下几种策略来处理多余的任务：

丢弃任务：直接丢弃任务。
等待任务空间：任务等待线程池有空余的线程来处理。
将任务存储到 Redis：将任务存储到 Redis 队列中，待线程池有空闲线程时再取出处理。

如果你使用的是 Spring 框架，并且希望在任务队列满时使用 RedisTemplate 将任务存储到 Redis 中，可以按照以下步骤进行实现：

关键步骤：

设置线程池：使用 ThreadPoolExecutor 配置线程池，并指定 RejectedExecutionHandler 来处理任务拒绝策略。
将任务存储到 Redis：当线程池队列满时，将任务存储到 Redis 队列中，待线程池有空闲时再取出处理。

示例代码

1. 创建 `ThreadPoolExecutor` 和自定义的任务拒绝策略：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import java.util.concurrent.*;@Service
public class TaskService {private static final String TASK_QUEUE = "taskQueue";// 定义线程池大小和队列容量private static final int THREAD_POOL_SIZE = 5;private static final int QUEUE_CAPACITY = 10;// 注入 RedisTemplate 用于操作 Redis@Autowiredprivate RedisTemplate<String, String> redisTemplate;private ExecutorService executorService;public TaskService() {// 创建线程池，设置自定义的任务拒绝策略executorService = new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE, 0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(QUEUE_CAPACITY), new ThreadPoolExecutor.CallerRunsPolicy());}// 提交任务到线程池public void submitTask(Runnable task) {try {executorService.submit(task);} catch (RejectedExecutionException e) {// 如果任务队列满了，将任务存储到 RedisSystem.out.println("线程池任务队列已满，将任务保存到 Redis...");redisTemplate.opsForList().leftPush(TASK_QUEUE, task.toString());}}// 从 Redis 中获取任务并执行public void processRedisTasks() {while (true) {// 从 Redis 中取出任务String taskFromRedis = redisTemplate.opsForList().rightPop(TASK_QUEUE);if (taskFromRedis != null) {// 执行从 Redis 取出的任务System.out.println("从 Redis 获取到任务：" + taskFromRedis);executorService.submit(() -> {System.out.println(Thread.currentThread().getName() + " 执行任务: " + taskFromRedis);try {// 模拟任务执行时间Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}try {// 暂停一段时间再去 Redis 获取任务Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}}
}

2. 创建任务并提交：

public class Task implements Runnable {private String taskName;public Task(String taskName) {this.taskName = taskName;}@Overridepublic void run() {// 任务执行逻辑System.out.println("任务: " + taskName + " 被执行...");try {Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}@Overridepublic String toString() {return taskName;}
}

3. 使用 `TaskService` 提交任务并启动 Redis 任务处理：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class Application implements CommandLineRunner {@Autowiredprivate TaskService taskService;public static void main(String[] args) {SpringApplication.run(Application.class, args);}@Overridepublic void run(String... args) throws Exception {// 模拟任务提交for (int i = 0; i < 20; i++) {taskService.submitTask(new Task("任务 " + i));}// 启动一个线程来处理从 Redis 获取的任务new Thread(() -> taskService.processRedisTasks()).start();}
}

关键点：

线程池配置：通过 ThreadPoolExecutor 设置线程池大小和任务队列的容量，使用 CallerRunsPolicy 作为拒绝策略，当队列满时，当前线程将会执行任务。
任务存储到 Redis：当线程池队列满时，将任务的标识（或者可以是任务本身）存储到 Redis 的队列中。RedisTemplate 提供了简洁的 API 用于操作 Redis 数据。
从 Redis 获取任务：在独立的线程中定时从 Redis 获取任务并提交到线程池中执行。