springboot线程池统一线程定义和使用

川长思鸟来 2023-01-20 09:43 23阅读 0赞

1.我们都知道spring只是为我们简单的处理线程池，每次用到线程总会new 一个新的线程，效率不高，所以我们需要自定义一个线程池。

2.自定义线程池有两种方法，第一种自定义线程池然后使用自己的自定义的，第二种重写spring默认的线程池，然后使用自己重写过的线程池

### 文章目录 ###

*   *  一：重写spring默认的线程池
     *  二：自定义线程池配置类
     *  三：使用
     *  四：工具类和注解等

## 一：重写spring默认的线程池 ##

import lombok.Data;
    import lombok.EqualsAndHashCode;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.util.ObjectUtils;
    import org.springframework.util.concurrent.ListenableFuture;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.Future;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /**
     * @author yzd
     * @Description: 可用线程池配置
     * <p>
     *     显示线程执行情况：任务总数、已完成数、活跃线程数，队列大小信息
     * </p>
     * @create 2021-04-29 9:28
     **/
    @Data
    @EqualsAndHashCode(callSuper = true)
    @Slf4j
    @Primary
    public class VisibleThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
        
    
        private static final long serialVersionUID = 8782155982988308307L;
    
        private void showThreadPoolInfo(String prefix) {
        
            ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();
    
            if (ObjectUtils.isEmpty(threadPoolExecutor)) {
        
                return;
            }
    
            log.info("{}, {},taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]",
                    this.getThreadNamePrefix(),
                    prefix,
                    threadPoolExecutor.getTaskCount(),
                    threadPoolExecutor.getCompletedTaskCount(),
                    threadPoolExecutor.getActiveCount(),
                    threadPoolExecutor.getQueue().size());
        }
    
        @Override
        public void execute(Runnable task) {
        
            showThreadPoolInfo("1. do execute");
            super.execute(task);
        }
    
        @Override
        public void execute(Runnable task, long startTimeout) {
        
            showThreadPoolInfo("2. do execute timeout");
            super.execute(task, startTimeout);
        }
    
        @Override
        public Future<?> submit(Runnable task) {
        
            showThreadPoolInfo("1. do submit");
            return super.submit(task);
        }
    
        @Override
        public <T> Future<T> submit(Callable<T> task) {
        
            showThreadPoolInfo("2. do submit");
            return super.submit(task);
        }
    
        @Override
        public ListenableFuture<?> submitListenable(Runnable task) {
        
            showThreadPoolInfo("1. do submitListenable");
            return super.submitListenable(task);
        }
    
        @Override
        public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
        
            showThreadPoolInfo("2. do submitListenable");
            return super.submitListenable(task);
        }
    }

## 二：自定义线程池配置类 ##

import lombok.extern.slf4j.Slf4j;
    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;
    
    /**
     * @author yzd
     * @Description: 线程池配置
     * @create 2021-04-29 9:37
     * <p>注意：使用注解 Async  开启异步 必须指定参数<>taskExecutor</> 如果不添加这个bean名称，会使用默认的线程池
     * </p>
     **/
    @Configuration
    @EnableAsync
    @Slf4j
    public class ExecutorConfig {
        
    
        @Bean
        @Primary
        public ThreadPoolTaskExecutor asyncServiceExecutor() {
        
            log.info("start asyncServiceExecutor");
            //使用visibleThreadPoolTaskExecutor
            ThreadPoolTaskExecutor executor = new VisibleThreadPoolTaskExecutor();
            // 通过Runtime方法来获取当前服务器cpu内核，根据cpu内核来创建核心线程数和最大线程数
            int availableProcessors = Runtime.getRuntime().availableProcessors();
            /**
             * 配置线程个数
             如果是CPU密集型任务，那么线程池的线程个数应该尽量少一些，一般为CPU的个数+1条线程(大量计算)
             如果是IO密集型任务，那么线程池的线程可以放的很大，如2*CPU的个数(IO操作)
             */
            executor.setCorePoolSize(availableProcessors * 2);
            // 允许线程池超时
            executor.setAllowCoreThreadTimeOut(true);
            //配置最大线程数
            executor.setMaxPoolSize(availableProcessors * 4);
            // 空闲存活时间
            executor.setKeepAliveSeconds(60);
            // 设置 等待终止秒数
            executor.setAwaitTerminationSeconds(60);
            //配置队列大小
            executor.setQueueCapacity(availableProcessors * 100);
            //配置线程池中的线程的名称前缀
            executor.setThreadNamePrefix("async-service-");
    
            // rejection-policy：当pool已经达到max size的时候，如何处理新任务
            // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            //执行初始化
            executor.initialize();
            return executor;
        }
    
    }

## 三：使用 ##

/**
     * @author yzd
     * @Description: 系统日志 aop
     * @create 2021-04-14 11:03
     **/
    @Slf4j
    @Aspect
    @Component
    public class LogAspect {
        
    
        @Autowired
        private ThreadPoolTaskExecutor asyncServiceExecutor;
    
        /**
         * 定义切入点
         */
        @Pointcut("execution(public * com.zbwd.outline.controller.*.*(..))")
        private void controllerAspect() {
        
        }
    
        @Around("controllerAspect()")
        public Object around(ProceedingJoinPoint pjp) throws Throwable {
        
    
            // 拦截目标
            Object target = pjp.getTarget();
            // 拦截方法
            String methodName = pjp.getSignature().getName();
    
            // 拦截参数
            Object[] args = pjp.getArgs();
    
            String description = String.format("method=%s, args=%s", methodName, Arrays.toString(args));
    
            Signature signature = pjp.getSignature();
    
            MethodSignature methodSignature = null;
    
            if (signature instanceof MethodSignature) {
        
                methodSignature = (MethodSignature) signature;
            }
    
            OperateLog operateLog = buildOperateLog();
    
            Object object;
            try {
        
    
                if (methodSignature == null) {
        
                    log.warn("获取方法签名失败");
                    return null;
                }
    
                Class<?>[] parameterTypes = methodSignature.getMethod().getParameterTypes();
    
                Method method = target.getClass().getMethod(methodName, parameterTypes);
    
                if (Objects.isNull(method)) {
        
                    return pjp.proceed();
                }
    
                if (Objects.isNull(method.getAnnotation(Log.class))) {
        
                    return pjp.proceed();
                }
                Log log = method.getAnnotation(Log.class);
                String logModule = log.logEnum().getModule();
                String logMethod = log.logEnum().getMethod();
                String title = log.logEnum().getTitle();
                operateLog.setModule(logModule);
                operateLog.setMethod(logMethod);
                operateLog.setStart(LocalDateTime.now());
                operateLog.setTitle(title);
    
                object = pjp.proceed();
    
                operateLog.setEnd(LocalDateTime.now());
            } catch (Exception exception) {
        
                log.error("获取日志失败啦：Description = {}", description, exception.getMessage());
                operateLog.setStatus(0);
                operateLog.setEnd(LocalDateTime.now());
                throw exception;
            } finally {
        
                if (!ObjectUtils.isEmpty(operateLog.getMethod())) {
        
                    this.runAsync(operateLog);
                }
            }
            return object;
        }
    
        private OperateLog buildOperateLog() {
        
            OperateLog operateLog = new OperateLog();
            HttpServletRequest request = ((ServletRequestAttributes) Objects.requireNonNull(RequestContextHolder.getRequestAttributes())).getRequest();
            operateLog.setIp(IpUtils.getIpAddr(request));
            return operateLog;
        }
    
    
        private void runAsync(OperateLog operateLog) {
        
            // 异步 CompletableFuture
            asyncServiceExecutor.execute(() -> {
        
                        // 日志存储
            );
        }
    }

## 四：工具类和注解等 ##

若需要具体的日志记录类和工具类等，如下：

*  ip工具类

import javax.servlet.http.HttpServletRequest;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.LineNumberReader;
    import java.net.InetAddress;
    import java.net.UnknownHostException;
    
    /**
     @Description: ip地址获取工具
     @author yzd
     @create 2021-04-14 10:59
     **/
    public class IpUtils {
        
    
        public static final String UN_KNOWN = "unknown";
        public static final String LOCAL = "127.0.0.1";
        public static final String LOCAL_STR = "0:0:0:0:0:0:0:1";
    
        /**
         * 获取当前网络ip
         * @param request
         * @return
         */
        public static String getIpAddr(HttpServletRequest request){
        
            String ipAddress = request.getHeader("x-forwarded-for");
            if (ipAddress == null || ipAddress.length() == 0 || UN_KNOWN.equalsIgnoreCase(ipAddress)) {
        
                ipAddress = request.getHeader("Proxy-Client-IP");
            }
            if (ipAddress == null || ipAddress.length() == 0 || UN_KNOWN.equalsIgnoreCase(ipAddress)) {
        
                ipAddress = request.getHeader("WL-Proxy-Client-IP");
            }
            if (ipAddress == null || ipAddress.length() == 0 || UN_KNOWN.equalsIgnoreCase(ipAddress)) {
        
                ipAddress = request.getRemoteAddr();
                if (LOCAL.equals(ipAddress) || LOCAL_STR.equals(ipAddress)) {
        
                    //根据网卡取本机配置的IP
                    InetAddress inet=null;
                    try {
        
                        inet = InetAddress.getLocalHost();
                    } catch (UnknownHostException e) {
        
                        e.printStackTrace();
                    }
                    ipAddress= inet.getHostAddress();
                }
            }
            //对于通过多个代理的情况，第一个IP为客户端真实IP,多个IP按照','分割
            int i = 15;
            if (ipAddress != null && ipAddress.length() > i) {
        
                String str = ",";
                if (ipAddress.indexOf(str) > 0) {
        
                    ipAddress = ipAddress.substring(0,ipAddress.indexOf(","));
                }
            }
            return ipAddress;
        }
    
        /**
         * 获得MAC地址
         * @param ip
         * @return
         */
        public static String getMacAddress(String ip){
        
            String str = "";
            String macAddress = "";
            try {
        
                Process p = Runtime.getRuntime().exec("nbtstat -A " + ip);
                InputStreamReader ir = new InputStreamReader(p.getInputStream());
                LineNumberReader input = new LineNumberReader(ir);
                int j = 100;
                for (int i = 1; i < j; i++) {
        
                    str = input.readLine();
                    if (str != null) {
        
                        if (str.indexOf("MAC Address") > 1) {
        
                            macAddress = str.substring(str.indexOf("MAC Address") + 14, str.length());
                            break;
                        }
                    }
                }
            } catch (IOException e) {
        
                e.printStackTrace(System.out);
            }
            return macAddress;
        }
    }

*  日志注解

import com.zbwd.outline.constant.LogEnum;
    
    import java.lang.annotation.*;
    
    /**
     @Description: 日志注解
     @author yzd
     @create 2021-04-14 10:58
     **/
    @Documented
    @Retention(RetentionPolicy.RUNTIME)
    @Target(value = {
        ElementType.PARAMETER, ElementType.METHOD})
    public @interface Log {
        
    
        LogEnum logEnum() ;
    
    }

*  日志枚举

import com.fasterxml.jackson.annotation.JsonFormat;
    import lombok.Getter;
    import lombok.ToString;
    
    /**
     @Description: 日志 枚举
     @author yzd
     @create 2021-04-19 9:18
     **/
    @JsonFormat(shape = JsonFormat.Shape.OBJECT)
    @Getter
    @ToString
    public enum  LogEnum {
        
    
        /**
         * 登录
         */
        USER_LOGIN("user","login","登录"),
    
        /**
         * 注销
         */
        USER_LOGOUT("user","logout","注销"),
    
        /**
         * 修改密码
         */
        USER_MODIFY_PASSWORD("user","modifyPassword","修改密码"),
        ;
    
        /**
         * 业务模块
         */
        private  String module;
    
        /**
         * 业务方法
         */
        private  String method;
    
        /**
         * 标题/描述/简介
         */
        private  String title;
    
    
        LogEnum(String module, String method, String title) {
        
            this.module = module;
            this.method = method;
            this.title = title;
        }
    }