【学习总结】SpringBoot中使用单例模式+ScheduledExecutorService实现异步多线程任务(若依源码学习)

最近在学习若依这个开源项目，发现他记录登录日志的时候使用了异步线程去记录日志，觉得这个方案也挺不错的，在此学习记录下来，以后在工作中也能提供一种思路，其他小伙伴如果有觉得不错的方案也可以在评论区里留言，大家一起探讨一下🍭 

若依源码

一、相关工具类

我们一步步看，先把相关的工具类代码给大家贴出来

1、Threads工具类

/**
 * 线程相关工具类.
 * 
 * @author ruoyi
 */
public class Threads
{
    private static final Logger logger = LoggerFactory.getLogger(Threads.class);
 
    /**
     * sleep等待,单位为毫秒
     */
    public static void sleep(long milliseconds)
    {
        try
        {
            Thread.sleep(milliseconds);
        }
        catch (InterruptedException e)
        {
            return;
        }
    }
 
    /**
     * 停止线程池
     * 先使用shutdown, 停止接收新任务并尝试完成所有已存在任务.
     * 如果超时, 则调用shutdownNow, 取消在workQueue中Pending的任务,并中断所有阻塞函数.
     * 如果仍然超時，則強制退出.
     * 另对在shutdown时线程本身被调用中断做了处理.
     */
    public static void shutdownAndAwaitTermination(ExecutorService pool)
    {
        if (pool != null && !pool.isShutdown())
        {
            pool.shutdown();
            try
            {
                if (!pool.awaitTermination(120, TimeUnit.SECONDS))
                {
                    pool.shutdownNow();
                    if (!pool.awaitTermination(120, TimeUnit.SECONDS))
                    {
                        logger.info("Pool did not terminate");
                    }
                }
            }
            catch (InterruptedException ie)
            {
                pool.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
 
    /**
     * 打印线程异常信息
     */
    public static void printException(Runnable r, Throwable t)
    {
        if (t == null && r instanceof Future)
        {
            try
            {
                Future future = (Future) r;
                if (future.isDone())
                {
                    future.get();
                }
            }
            catch (CancellationException ce)
            {
                t = ce;
            }
            catch (ExecutionException ee)
            {
                t = ee.getCause();
            }
            catch (InterruptedException ie)
            {
                Thread.currentThread().interrupt();
            }
        }
        if (t != null)
        {
            logger.error(t.getMessage(), t);
        }
    }
}

这个工具类包含了三个方法：

•  sleep(long milliseconds)：这个方法比较简单，就是用来睡眠线程的

• shutdownAndAwaitTermination(ExecutorService pool)（重点）：该方法用于优雅地关闭一个 ExecutorService 并等待其终止。

public static void shutdownAndAwaitTermination(ExecutorService pool)
    {
        if (pool != null && !pool.isShutdown()) //确保传入的 ExecutorService 不为 null 并且尚未被关闭。
        {
            /*
                调用 shutdown 方法，这将启动线程池的关闭序列。注意，shutdown 并不会立即停                
              止所有正在执行的任务，也不会阻止新任务的提交。它只会使 ExecutorService 不再接受 
              新任务，但会等待已提交的任务完成。
            */
            pool.shutdown(); 
            try
            {
                if (!pool.awaitTermination(120, TimeUnit.SECONDS))//等待线程池在指定的时间                    
         内（这里是120秒）完成所有任务并关闭。如果线程池在该时间内没有关闭，则会进入 if 语句块
                {
                    pool.shutdownNow(); //强制停止所有正在执行的任务，它并不能保证所有的任务都会被停止
                    if (!pool.awaitTermination(120, TimeUnit.SECONDS))//在强制关闭线程池后，再次等待120秒以查看它是否已关闭。
                    {
                        logger.info("Pool did not terminate");//如果仍未关闭，则记录一条日志信息。
                    }
                }
            }
            catch (InterruptedException ie)
            {
                /*
                     如果在等待线程池关闭时被中断（例如，由于另一个线程调用了当前线程的     
                  interrupt 方法），则再次调用 shutdownNow 强制关闭线程池，并重新设置当前线 
                  程的中断状态。
                */
                pool.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }

• printException(Runnable r, Throwable t) ：这个方法就是用于记录异常信息，写的也是很优雅。

2、SpringUtils工具类

/**
 * spring工具类 方便在非spring管理环境中获取bean
 * 
 * @author ruoyi
 */
@Component
public final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware 
{
    /** Spring应用上下文环境 */
    private static ConfigurableListableBeanFactory beanFactory;
 
    private static ApplicationContext applicationContext;
 
    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException 
    {
        SpringUtils.beanFactory = beanFactory;
    }
 
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException 
    {
        SpringUtils.applicationContext = applicationContext;
    }
 
    /**
     * 获取对象
     *
     * @param name
     * @return Object 一个以所给名字注册的bean的实例
     * @throws BeansException
     *
     */
    @SuppressWarnings("unchecked")
    public static  T getBean(String name) throws BeansException
    {
        return (T) beanFactory.getBean(name);
    }
 
    /**
     * 获取类型为requiredType的对象
     *
     * @param clz
     * @return
     * @throws BeansException
     *
     */
    public static  T getBean(Class clz) throws BeansException
    {
        T result = (T) beanFactory.getBean(clz);
        return result;
    }
 
    /**
     * 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
     *
     * @param name
     * @return boolean
     */
    public static boolean containsBean(String name)
    {
        return beanFactory.containsBean(name);
    }
 
    /**
     * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
     *
     * @param name
     * @return boolean
     * @throws NoSuchBeanDefinitionException
     *
     */
    public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException
    {
        return beanFactory.isSingleton(name);
    }
 
    /**
     * @param name
     * @return Class 注册对象的类型
     * @throws NoSuchBeanDefinitionException
     *
     */
    public static Class getType(String name) throws NoSuchBeanDefinitionException
    {
        return beanFactory.getType(name);
    }
 
    /**
     * 如果给定的bean名字在bean定义中有别名，则返回这些别名
     *
     * @param name
     * @return
     * @throws NoSuchBeanDefinitionException
     *
     */
    public static String[] getAliases(String name) throws NoSuchBeanDefinitionException
    {
        return beanFactory.getAliases(name);
    }
 
    /**
     * 获取aop代理对象
     * 
     * @param invoker
     * @return
     */
    @SuppressWarnings("unchecked")
    public static  T getAopProxy(T invoker)
    {
        return (T) AopContext.currentProxy();
    }
 
    /**
     * 获取当前的环境配置，无配置返回null
     *
     * @return 当前的环境配置
     */
    public static String[] getActiveProfiles()
    {
        return applicationContext.getEnvironment().getActiveProfiles();
    }
 
    /**
     * 获取当前的环境配置，当有多个环境配置时，只获取第一个
     *
     * @return 当前的环境配置
     */
    public static String getActiveProfile()
    {
        final String[] activeProfiles = getActiveProfiles();
        return StringUtils.isNotEmpty(Arrays.toString(activeProfiles)) ? activeProfiles[0] : null;
    }
 
    /**
     * 获取配置文件中的值
     *
     * @param key 配置文件的key
     * @return 当前的配置文件的值
     *
     */
    public static String getRequiredProperty(String key)
    {
        return applicationContext.getEnvironment().getRequiredProperty(key);
    }
}

二、核心代码

1、异步任务管理器

/**
 * 异步任务管理器
 * 
 * @author ruoyi
 */
public class AsyncManager
{
    /**
     * 操作延迟10毫秒
     */
    private final int OPERATE_DELAY_TIME = 10;
 
    /**
     * 异步操作任务调度线程池
     */
    private ScheduledExecutorService executor = SpringUtils.getBean("scheduledExecutorService");
 
    /**
     * 单例模式
     */
    private AsyncManager(){}
 
    private static AsyncManager me = new AsyncManager();
 
    public static AsyncManager me()
    {
        return me;
    }
 
    /**
     * 执行任务
     * 
     * @param task 任务
     */
    public void execute(TimerTask task)
    {
        executor.schedule(task, OPERATE_DELAY_TIME, TimeUnit.MILLISECONDS);
    }
 
    /**
     * 停止任务线程池
     */
    public void shutdown()
    {
        Threads.shutdownAndAwaitTermination(executor);
    }
}

2、异步工厂

/**
 * 异步工厂（产生任务用）
 * 
 * @author ruoyi
 */
public class AsyncFactory
{
    private static final Logger sys_user_logger = LoggerFactory.getLogger("sys-user");
 
    /**
     * 记录登录信息
     * 
     * @param username 用户名
     * @param status 状态
     * @param message 消息
     * @param args 列表
     * @return 任务task
     */
    public static TimerTask recordLogininfor(final String username, final String status, final String message,
            final Object... args)
    {
        final UserAgent userAgent = UserAgent.parseUserAgentString(ServletUtils.getRequest().getHeader("User-Agent"));
        final String ip = IpUtils.getIpAddr();
        return new TimerTask()
        {
            @Override
            public void run()
            {
                String address = AddressUtils.getRealAddressByIP(ip);
                StringBuilder s = new StringBuilder();
                s.append(LogUtils.getBlock(ip));
                s.append(address);
                s.append(LogUtils.getBlock(username));
                s.append(LogUtils.getBlock(status));
                s.append(LogUtils.getBlock(message));
                // 打印信息到日志
                sys_user_logger.info(s.toString(), args);
                // 获取客户端操作系统
                String os = userAgent.getOperatingSystem().getName();
                // 获取客户端浏览器
                String browser = userAgent.getBrowser().getName();
                // 封装对象
                SysLogininfor logininfor = new SysLogininfor();
                logininfor.setUserName(username);
                logininfor.setIpaddr(ip);
                logininfor.setLoginLocation(address);
                logininfor.setBrowser(browser);
                logininfor.setOs(os);
                logininfor.setMsg(message);
                // 日志状态
                if (StringUtils.equalsAny(status, Constants.LOGIN_SUCCESS, Constants.LOGOUT, Constants.REGISTER))
                {
                    logininfor.setStatus(Constants.SUCCESS);
                }
                else if (Constants.LOGIN_FAIL.equals(status))
                {
                    logininfor.setStatus(Constants.FAIL);
                }
                // 插入数据
                SpringUtils.getBean(ISysLogininforService.class).insertLogininfor(logininfor);
            }
        };
    }
 
    /**
     * 操作日志记录
     * 
     * @param operLog 操作日志信息
     * @return 任务task
     */
    public static TimerTask recordOper(final SysOperLog operLog)
    {
        return new TimerTask()
        {
            @Override
            public void run()
            {
                // 远程查询操作地点
                operLog.setOperLocation(AddressUtils.getRealAddressByIP(operLog.getOperIp()));
                SpringUtils.getBean(ISysOperLogService.class).insertOperlog(operLog);
            }
        };
    }
}

3、线程关闭

@Component
public class ShutdownManager
{
    private static final Logger logger = LoggerFactory.getLogger("sys-user");
 
    @PreDestroy
    public void destroy()
    {
        shutdownAsyncManager();
    }
 
    /**
     * 停止异步执行任务
     */
    private void shutdownAsyncManager()
    {
        try
        {
            logger.info("====关闭后台任务任务线程池====");
            AsyncManager.me().shutdown();
        }
        catch (Exception e)
        {
            logger.error(e.getMessage(), e);
        }
    }
}

改造代码

1、改造异步任务管理器

这个管理器里面没有业务代码，我们稍微修改一下。

public class GlobalAsyncManager {
 
    //单例
    private static final GlobalAsyncManager instance = new GlobalAsyncManager();
 
    //延迟执行时间
    private final int OPERATOR_DELAY_TIME = 10;
 
    private ScheduledExecutorService executorService = SpringUtils.getBean("scheduledExecutorService");
 
    private GlobalAsyncManager(){
 
    }
 
    public static GlobalAsyncManager getInstance(){
        return  instance;
    }
 
    //执行任务
    public void executorTask(TimerTask task){
        executorService.schedule(task,OPERATOR_DELAY_TIME, TimeUnit.MILLISECONDS);
    }
 
    //停止任务线程池
    public void shutdown(){
        Threads.shutdownAndAwaitTermination(executorService);
    }
}

2、自定义异步工厂

/**
 * 异步工厂，产生任务用
 */
@Slf4j
public class AsyncTaskFactory {
    public static TimerTask recordLogToData(参数){
        return new TimerTask() {
            @Override
            public void run() {
                //日志操作
            }
        };
    }
}

使用

1、若依代码使用

/**
     * 登录验证
     * 
     * @param username 用户名
     * @param password 密码
     * @param code 验证码
     * @param uuid 唯一标识
     * @return 结果
     */
    public String login(String username, String password, String code, String uuid)
    {
        // 验证码校验
        validateCaptcha(username, code, uuid);
        // 登录前置校验
        loginPreCheck(username, password);
        // 用户验证
        Authentication authentication = null;
        try
        {
            UsernamePasswordAuthenticationToken authenticationToken = new UsernamePasswordAuthenticationToken(username, password);
            AuthenticationContextHolder.setContext(authenticationToken);
            // 该方法会去调用UserDetailsServiceImpl.loadUserByUsername
            authentication = authenticationManager.authenticate(authenticationToken);
        }
        catch (Exception e)
        {
            if (e instanceof BadCredentialsException)
            {
                AsyncManager.me().execute(AsyncFactory.recordLogininfor(username, Constants.LOGIN_FAIL, MessageUtils.message("user.password.not.match")));
                throw new UserPasswordNotMatchException();
            }
            else
            {
                AsyncManager.me().execute(AsyncFactory.recordLogininfor(username, Constants.LOGIN_FAIL, e.getMessage()));
                throw new ServiceException(e.getMessage());
            }
        }
        finally
        {
            AuthenticationContextHolder.clearContext();
        }
        AsyncManager.me().execute(AsyncFactory.recordLogininfor(username, Constants.LOGIN_SUCCESS, MessageUtils.message("user.login.success")));
        LoginUser loginUser = (LoginUser) authentication.getPrincipal();
        recordLoginInfo(loginUser.getUserId());
        // 生成token
        return tokenService.createToken(loginUser);
    }

2、自定义代码使用

    @Test
    public void recordLog() throws InterruptedException {
       //为了验证其是异步效果，这里模拟线程休眠并记录时间
        Thread.sleep(1000);         
 
GlobalAsyncManager.getInstance().executorTask(AsyncTaskFactory.recordLogToData(LocalDateTime.now());
        Thread.sleep(3000);
        System.out.println(LocalDateTime.now());
    }