JDK自带的延迟队列-DelayQueue

package com.example.delayedqueuedemo.service; import com.example.delayedqueuedemo.task.DelayedTask; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionException; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.DefaultTransactionDefinition; import javax.annotation.PostConstruct; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.DelayQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @Slf4j @Service public class DelayedService { private ThreadPoolExecutor takeTaskPool = new ThreadPoolExecutor(1, 1, 120, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1)); private ThreadPoolExecutor executeTaskPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(), 120, TimeUnit.SECONDS, new ArrayBlockingQueue<>(Runtime.getRuntime().availableProcessors()), new ThreadPoolExecutor.CallerRunsPolicy()); private DelayQueue<DelayedTask> delayQueue = new DelayQueue<>(); @Autowired private PlatformTransactionManager transactionManager; /** * 生命周期回调 */ @PostConstruct public void postConstruct() { // 开启线程池，避免死循环阻塞主程序 takeTaskPool.execute(new Runnable() { @Override public void run() { // 如果队列中没有到期的元素，take会阻塞等待 while (true) { try { DelayedTask task = delayQueue.take(); // 开启线程池，避免业务执行阻塞队列获取消息，核心线程数根据电脑cpu核数而定 executeTaskPool.execute(new Runnable() { @Override public void run() { // 开启事务 TransactionDefinition definition = new DefaultTransactionDefinition(); TransactionStatus transaction = transactionManager.getTransaction(definition); try { // 业务逻辑 log.info("获取延迟任务：task：{}", task.getData()); // 提交事务 transactionManager.commit(transaction); } catch (TransactionException e) { // 回滚事务 transactionManager.rollback(transaction); } } }); } catch (InterruptedException e) { e.printStackTrace(); } } } }); } @Transactional public void delayed(String data, Long expireTime) { // 1.初始化延迟队列 DelayedTask task = new DelayedTask(data, expireTime); // 2.向队列中添加延迟执行的任务 delayQueue.put(task); // 3.尝试执行任务 log.info("消息发送完成"); } }