基于自定义线程池手写一个异步任务管理器

简介: 我们在后端执行某些耗时逻辑操作时往往会导致长时间的线程阻塞,在这种情况之下,我们往往会引一条异步线程去处理这些异步任务,如果每次都创建新的线程来处理这些任务,不仅会增加代码冗余,还可能造成线程管理混乱,影响系统性能。在我们的Spring框架中是自带异步任务处理机制的,比如我们使用@Async 注解可以处理一些简单的异步任务,但这样确实无法精确去控制线程池资源,也无法灵活去管理任务调度,由此,我们可以去自行设计一个高效的自定义异步任务管理器去统一调度处理我们的自定义任务。

 我们在后端执行某些耗时逻辑操作时往往会导致长时间的线程阻塞,在这种情况之下,我们往往会引一条异步线程去处理这些异步任务,如果每次都创建新的线程来处理这些任务,不仅会增加代码冗余,还可能造成线程管理混乱,影响系统性能。在我们的Spring框架中是自带异步任务处理机制的,比如我们使用@Async 注解可以处理一些简单的异步任务,但这样确实无法精确去控制线程池资源,也无法灵活去管理任务调度,由此,我们可以去自行设计一个高效的自定义异步任务管理器去统一调度处理我们的自定义任务。

1.前置配置

自定义线程池,并将其注册到IOC容器中

/**
 * 自定义线程池配置
 * @Author GuihaoLv
 **/
@Configuration
public class ThreadPoolConfig
{
    // 核心线程池大小
    private int corePoolSize = 50;
    // 最大可创建的线程数
    private int maxPoolSize = 200;
    // 队列最大长度
    private int queueCapacity = 1000;
    // 线程池维护线程所允许的空闲时间
    private int keepAliveSeconds = 300;
    /**
     * 通用任务线程池
     * @return
     */
    @Bean(name = "threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor()
    {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setMaxPoolSize(maxPoolSize);
        executor.setCorePoolSize(corePoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setKeepAliveSeconds(keepAliveSeconds);
        // 线程池对拒绝任务(无线程可用)的处理策略
        //当线程池满了,新任务无法加入时,CallerRunsPolicy 让提交任务的线程(即调用方线程)直接执行该任务,
        // 而不是丢弃或抛出异常,从而保证任务不会丢失。
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
    /**
     * 执行周期性或定时任务
     */
    @Bean(name = "scheduledExecutorService")
    protected ScheduledExecutorService scheduledExecutorService()
    {
        //这里没有最大线程数的概念,所有线程都属于核心线程。
        return new ScheduledThreadPoolExecutor(corePoolSize,
                new BasicThreadFactory.Builder().namingPattern("schedule-pool-%d") //设置线程名称,方便排查日志。
                        .daemon(true).build(), //daemon(true) 使线程池中的线程成为 守护线程,即 JVM 退出时不会阻止进程终止。
                new ThreadPoolExecutor.CallerRunsPolicy()) //使用 CallerRunsPolicy,避免任务丢失。
        {
            //任务执行完毕后,调用 Threads.printException(r, t),捕获并记录异常,确保线程池不会因为未捕获的异常而崩溃。
            @Override
            protected void afterExecute(Runnable r, Throwable t)
            {
                super.afterExecute(r, t);
                Threads.printException(r, t);
            }
        };
    }
}

image.gif

2.异步任务管理器配置

/**
 * 异步任务管理器
 * AsyncManager 是 整个异步任务调度的核心,它提供了 任务执行、调度和管理。
 * @Author GuihaoLv
 */
public class AsyncManager
{
    /**
     * 操作延迟10毫秒
     */
    private final int OPERATE_DELAY_TIME = 10;
    /**
     * 异步操作任务调度线程池
     * executor 采用 ScheduledExecutorService 线程池,可以 定时执行异步任务,提高并发能力
     */
    private ScheduledExecutorService executor = SpringUtils.getBean("scheduledExecutorService");
    /**
     * 单例模式
     * 采用 单例模式,确保全局只有一个 AsyncManager 实例,保证任务调度统一管理。
     创建单例对象
     */
    private AsyncManager(){}
    //创建异步任务管理器的静态对象
    private static AsyncManager me = new AsyncManager();
    public static AsyncManager me()
    {
        return me;
    }
    /**
     * 使用调度线程池执行任务
     * @param task 任务
     */
    //TimerTask是Java编程语言中的一个抽象类,通常用于安排将来某个时间执行的任务,或者以固定的速率重复执行的任务。
    // 它是与Timer类一起使用的,Timer负责管理和调度这些任务。
    public void execute(TimerTask task)
    {
        executor.schedule(task, OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS);
    }
    /**
     * 停止任务线程池
     */
    public void shutdown()
    {
        //优雅关闭线程池
        Threads.shutdownAndAwaitTermination(executor);
    }
}

image.gif

/**
 * 确保应用退出时能关闭后台线程
 * @Author GuihaoLv
 */
@Component
public class ShutdownManager
{
    private static final Logger logger = LoggerFactory.getLogger("sys-user");
    @PreDestroy //它用于在 Spring Bean 被销毁前执行清理逻辑。
    public void destroy()
    {
        shutdownAsyncManager();
        HttpUtils.shutdown();
    }
     /**
     * 停止异步执行任务
     */
    private void shutdownAsyncManager()
    {
        try
        {
            logger.info("====关闭后台任务任务线程池====");
            AsyncManager.me().shutdown();
        }
        catch (Exception e)
        {
            logger.error(e.getMessage(), e);
        }
    }
}

image.gif

3.异步工厂配置,将异步任务的启动逻辑都扔到异步工厂中处理

/**
 * 异步工厂(产生任务用)
 * AsyncFactory 主要用于创建异步任务,它相当于一个 "任务工厂",
 * 可以根据不同的需求创建不同的任务(如记录用户登录信息、记录操作日志)。
 *
 * @Author GuihaoLv
 */
public class AsyncFactory
{
    private static final Logger sys_user_logger = LoggerFactory.getLogger("sys-user");
    // 定义任务:计算热点文章
    public static TimerTask calculateHotArticlesTask() {
        return new TimerTask() {
            @Override
            public void run() {
                try {
                    // 获取 HotArticleAsycTask 实例
                    HotArticleAsycTask hotArticleAsycTask = SpringUtils.getBean(HotArticleAsycTask.class);
                    // 执行热点文章计算任务
                    hotArticleAsycTask.calculateHotArticles();
                } catch (Exception e) {
                    sys_user_logger.error("计算热点文章失败", e);
                }
            }
        };
    }
    // 定义任务:定时热搜清理
    public static TimerTask cleanupOldKeywords() {
        return new TimerTask() {
            @Override
            public void run() {
                try {
                    // 获取 HotArticleAsycTask 实例
                    HotSearchCleanupTask hotArticleAsycTask = SpringUtils.getBean(HotSearchCleanupTask.class);
                    // 执行热点文章计算任务
                    hotArticleAsycTask.cleanupOldKeywords();
                } catch (Exception e) {
                    sys_user_logger.error("清理热搜失败", e);
                }
            }
        };
    }
}

image.gif

4.调用异步工厂中的逻辑

/**
 * 热点文章实时计算
 * @Author GuihaoLv
 */
@Component
public class HotArticleAsycTask {
    @Autowired
    private StringRedisTemplate redisTemplate;
    private static final String LIKES_KEY = "blog:likes";
    private static final String FAVORITES_KEY = "blog:favorites";
    private static final String HOT_ARTICLES_KEY = "blog:hot"; // 存热点文章
    public void calculateHotArticles() {
        Set<String> blogIds = redisTemplate.opsForZSet().range(LIKES_KEY, 0, -1);
        if (blogIds == null || blogIds.isEmpty()) return;
        for (String blogId : blogIds) {
            Double likes = redisTemplate.opsForZSet().score(LIKES_KEY, blogId);
            Double favorites = redisTemplate.opsForZSet().score(FAVORITES_KEY, blogId);
            // 计算热度
            double hotScore = (likes != null ? likes * 5.0 : 0) +
                    (favorites != null ? favorites * 8.0 : 0);
            // 热度超过阈值 500,加入热点文章
            if (hotScore >= 500) {
                redisTemplate.opsForZSet().add(HOT_ARTICLES_KEY, blogId, hotScore);
                System.out.println("🔥 文章 " + blogId + " 进入热点榜,热度:" + hotScore);
            }
        }
    }
     /**
      * 每 5 分钟执行一次的定时任务
      */
     @Scheduled(fixedRate = 100000) // 每 5 分钟执行一次
     public void scheduleHotArticlesCalculation() {
         // 将任务交给异步任务管理器执行
         AsyncManager.me().execute(AsyncFactory.calculateHotArticlesTask());
     }
}

image.gif


相关文章
|
4月前
|
消息中间件 JavaScript 前端开发
详解事件循环与浏览器渲染机制
摘要:浏览器采用多进程架构,渲染主线程通过事件循环机制处理HTML解析、样式计算、布局等任务。异步机制避免主线程阻塞,任务按优先级在微队列、交互队列等不同队列中调度。JS执行会阻碍渲染,因其与渲染任务共享主线程。渲染流程包含解析、样式计算、布局、分层等阶段,最终由合成线程和GPU完成绘制。transform效率高因其仅影响合成阶段,不涉及主线程。reflow是布局重计算,repaint是绘制指令更新,两者均影响性能。
|
4月前
|
消息中间件 Dubbo Java
深入剖析RocketMQ2-实战案例
本文介绍了一个基于电商场景的订单和支付系统实现方案。系统采用SpringBoot+Dubbo+Zookeeper+RocketMQ技术栈,重点解决分布式系统中的数据一致性问题。在订单模块,通过预订单生成、库存扣减、优惠券使用、余额扣减等步骤完成下单流程,并利用RocketMQ实现失败补偿机制。支付模块处理第三方支付回调,通过消息队列异步更新订单状态。系统通过分布式事务保证数据一致性,采用线程池优化消息发送性能,并提供完整的测试方案验证系统功能。本文详细阐述了技术架构、数据库设计、核心业务流程以及异常处理机制
|
4月前
|
机器学习/深度学习 存储 人工智能
大模型应用开发1-认识大模型
摘要: 本文系统介绍了大模型的基础概念、本地部署及API调用方法。首先阐述了AI及神经网络的基本原理,重点解析了Transformer架构及其在大语言模型(LLM)中的应用。其次详细对比了三种模型使用方案(开放API/云部署/本地部署)的优缺点,并以Ollama为例演示了本地部署流程,包括模型管理、交互指令和GPU加速配置。最后说明了大模型API调用规范,列举了主流大模型产品及其应用场景,强调大模型在自然语言处理、内容生成等领域的优势,以及与传统编程结合开发智能应用的可能性。全文涵盖技术原理到实践操作,为大
|
4月前
|
消息中间件 存储 Kafka
详解Kafka1-基础使用
Kafka消息队列技术指南 本文系统介绍了Kafka消息队列的核心概念与应用实践。主要内容包括: 消息队列基础 两种模型:生产者-消费者模型和发布-订阅模型 应用场景:异步处理、系统解耦、流量削峰等 Kafka核心架构 重要组件:Broker、Zookeeper、Topic、Partition等 消费者组机制实现负载均衡 消息存储与分区策略 实践操作 集群搭建与环境配置 基准测试方法与性能指标 Java API编程示例(生产者/消费者) 事务编程实现原子操作 高级特性 生产者幂等性原理与配置 事务API与隔
|
4月前
|
前端开发 Java Maven
MinIO的预签名直传机制
我们传统使用MinIo做OSS对象存储的应用方式往往都是在后端配置与MinIO的连接和文件上传下载的相关接口,然后我们在前端调用这些接口完成文件的上传下载机制,但是,当并发量过大,频繁访问会对后端的并发往往会对服务器造成极大的压力,大文件传输场景下,服务器被迫承担数据中转的角色,既消耗大量带宽资源,又形成单点性能瓶颈。这时,我们引入了MinIO的一种预签名机制。
MinIO的预签名直传机制
|
4月前
|
存储 监控 前端开发
大文件上传下载处理方案-断点续传,秒传,分片,合并
本文介绍了大文件上传下载的断点续传技术方案。上传方面,通过前端将大文件分块(如5MB/块),后端使用MinIO存储分块并合并,实现断点续传和秒传功能。下载方面,采用Range请求分片下载,前端合并分片触发下载。技术要点包括:1)前端分块计算MD5;2)后端MinIO存储管理;3)分片校验与合并;4)进度监控和异常处理。该方案解决了大文件传输中断问题,提升用户体验,适用于视频等大文件传输场景,完整代码示例包含前后端实现。
|
4月前
|
安全 NoSQL Java
基于JWT+SpringSecurity整合一个单点认证授权机制
本文介绍了基于JWT和SpringSecurity的授权认证机制架构设计。系统采用RBAC权限模型,通过5张表描述用户-角色-权限关系。认证流程包含登录验证、IP检查、密码匹配等环节,使用JWT生成token并保存用户信息到Redis。授权部分利用@PreAuthorize注解和PermissionService实现权限校验,支持单权限、多权限及角色验证。整体架构通过过滤器链实现无状态认证,兼顾安全性和灵活性,为开发者提供了完整的认证授权解决方案。
|
4月前
|
消息中间件 存储 监控
详解Kafka2-进阶机制
本文摘要: 文章详细介绍了Kafka的分区与副本机制,包括生产者分区写入策略(轮询、随机、按key分配、自定义)、消费者组Rebalance机制及其影响,以及消费者分区分配策略(Range、RoundRobin、Sticky)。同时,阐述了Kafka的副本机制、数据存储形式(Segment结构)、消息不丢失机制(生产者ACK、消费者Offset管理)和数据积压问题。此外,还介绍了Kafka的数据清理策略(日志删除与压缩)、配额限速机制(Quotas)及监控工具Kafka-Eagle的使用方法。通过原理分析与
|
4月前
|
人工智能 JavaScript 数据挖掘
大模型应用开发4-MCP实战
MCP(模型上下文协议)是由Anthropic推出的开放标准,旨在解决AI互联领域中智能体与工具间的交互问题。该协议提供统一的数据访问方式,克服传统AI集成碎片化的局限性。MCP支持两种通信机制:stdio(本地进程通信)和SSE(远程服务调用)。其架构包含MCP Host(主应用)、MCP Client(请求转发)和MCP Server(工具执行)三部分。文章详细介绍了MCP的安装配置、工作原理,并通过一个本地舆情分析系统的开发案例,展示了如何实现客户端与服务端的交互。此外还介绍了CherryStudio
|
4月前
|
存储 算法 JavaScript
OAuth2协议与微信扫码登录实战
本文介绍了OAuth2协议及其在微信扫码登录中的应用实现。OAuth2是一种开放、安全的授权标准,支持授权码、密码等多种模式,其中授权码模式被微信扫码登录采用。文章详细解析了微信扫码登录流程:用户扫码授权→获取授权码→换取令牌→获取用户信息→完成登录。技术实现方面,使用JWT令牌解决认证服务与资源服务间的校验问题,并提供了完整的后端接口设计(包括获取授权URL、处理回调等)和前端测试页面代码。系统通过网关进行统一认证,各微服务自行处理授权,确保安全性的同时提高性能。整体方案实现了基于OAuth2协议的无缝第