<dependency>
	<groupId>org.redisson</groupId>
	<artifactId>redisson-spring-boot-starter</artifactId>
	<version>3.21.1</version>
</dependency>

# redis配置
spring:
  redis:
    database: 0
    host: 127.0.0.1
    password: redis@pass
    port: 6001

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync
public class ExecutorConfig {
    /**
     * 異步任務(wù)自定義線程池
     */
    @Bean(name = "taskExecutor")
    public ThreadPoolTaskExecutor asyncServiceExecutor() {
    	ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心線程數(shù)
        executor.setCorePoolSize(50);
        //配置最大線程數(shù)
        executor.setMaxPoolSize(500);
        //配置隊列大小
        executor.setQueueCapacity(300);
        //允許線程空閑時間
        executor.setKeepAliveSeconds(60);
        //配置線程池中的線程的名稱前綴
        executor.setThreadNamePrefix("taskExecutor-");
        // rejection-policy：當(dāng)pool已經(jīng)達到max size的時候，如何處理新任務(wù)
        // CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是有調(diào)用者所在的線程來執(zhí)行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //調(diào)用shutdown()方法時等待所有的任務(wù)完成后再關(guān)閉
        executor.setWaitForTasksToCompleteOnShutdown(true);
        //等待所有任務(wù)完成后的最大等待時間
		executor.setAwaitTerminationSeconds(60);
        return executor;
    }
}

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.Date;

@Service
public class AsyncService {

	@Async
	public void executeQueue(Object value) {
		System.out.println();
		System.out.println("當(dāng)前線程："+Thread.currentThread().getName());
		System.out.println("執(zhí)行任務(wù)："+value);

		//打印時間方便查看
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		System.out.println("執(zhí)行任務(wù)的時間："+sdf.format(new Date()));
		//自己的業(yè)務(wù)邏輯，可以根據(jù)id發(fā)送通知消息等
		//......
	}
}

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Service
public class TestService {

	@Resource
	private AsyncService asyncService;
	@Resource
	private ThreadPoolTaskExecutor executor;
	@Autowired
	private RedissonClient redissonClient;

	/**
	 * 添加延遲任務(wù)
	 */
	public void addQueue() {
		//獲取延遲隊列
		RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue("delayedQueue");
		RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
		for (int i = 1; i <= 10; i++) {
			long delayTime = 5+i; //延遲時間(秒)
//			long delayTime = 5; //這里時間統(tǒng)一，可以測試并發(fā)執(zhí)行
			delayedQueue.offer("延遲任務(wù)"+i, delayTime, TimeUnit.SECONDS);
		}
		//打印時間方便查看
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		System.out.println("添加任務(wù)的時間："+sdf.format(new Date()));
	}

	/**
	 *	服務(wù)啟動時加載，開始消費延遲隊列
	 */
	@PostConstruct
	public void consumer() {
		System.out.println("服務(wù)啟動時加載>>>>>>");
		//獲取延遲隊列
		RBlockingQueue<Object> delayedQueue = redissonClient.getBlockingQueue("delayedQueue");

		//啟用一個線程來消費這個延遲隊列
		executor.execute(() ->{
			while (true){
				try {
//					System.out.println("while中的線程："+Thread.currentThread().getName());
					//獲取延遲隊列中的任務(wù)
					Object value = delayedQueue.poll();
					if(value == null){
						//如果沒有任務(wù)就休眠1秒，休眠時間根據(jù)業(yè)務(wù)自己定義
						Thread.sleep(1000);	//這里休眠時間越短，誤差就越小
						continue;
					}
					//異步處理延遲隊列中的消息
					asyncService.executeQueue(value);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
	}
}

import com.test.service.TestService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/test")
public class TestController {

	@Autowired
	private TestService testService;

	/*
	 * 添加延遲任務(wù)
	 */
	@GetMapping(value = "/addQueue")
	public String addQueue() {
		testService.addQueue();
		return "success";
	}

}