springboot配置线程池-高并发场景

 

1.编写配置文件

package com.hbedu.search.hbsearch.utils;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

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.Executor;

import java.util.concurrent.ThreadPoolExecutor;


@Configuration

@EnableAsync

public class ExecutorConfig {



    private static final Logger logger = LoggerFactory.getLogger(ExecutorConfig.class);


    @Bean

    public Executor asyncServiceExecutor() {

        logger.info("start asyncServiceExecutor");

        //ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

        ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();

        //配置核心线程数

        executor.setCorePoolSize(5);

        //配置最大线程数

        executor.setMaxPoolSize(5);

        //配置队列大小

        executor.setQueueCapacity(99999);

        //配置线程池中的线程的名称前缀

        executor.setThreadNamePrefix("async-service-");

        // rejection-policy：当pool已经达到max size的时候，如何处理新任务

        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行

        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        //执行初始化

        executor.initialize();

        return executor;

    }


}

 

2.创建接口 AsyncService

package com.hbedu.search.hbsearch.service;

public interface AsyncService {

    /**

     * 执行异步任务

     */

    void executeAsync();

}

3.编写AsyncService实现

package com.hbedu.search.hbsearch.service.impl;

import com.hbedu.search.hbsearch.service.AsyncService;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Service;


@Service

public class AsyncServiceImpl implements AsyncService {



    private static final Logger logger = LoggerFactory.getLogger(AsyncServiceImpl.class);



    @Override

    @Async("asyncServiceExecutor")

    public void executeAsync() {

        logger.info("start executeAsync");

        try{

            Thread.sleep(1000);

        }catch(Exception e){

            e.printStackTrace();

        }

        logger.info("end executeAsync");

    }

}

 

4.编写controller

package com.hbedu.search.hbsearch.controller;



import com.hbedu.search.hbsearch.service.AsyncService;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;



@RestController

public class ExecutorController {

    private static final Logger logger = LoggerFactory.getLogger(ExecutorController.class);

    @Autowired

    private AsyncService asyncService;



    @RequestMapping("/ececutor")

    public String submit(){

        logger.info("start submit");


        //调用service层的任务

        asyncService.executeAsync();


        logger.info("end submit");

        return "success";

    }


}

 

5.启动服务调用接口查看 快速刷新（如果效果不明显，请增加线程睡眠时间），目的是复现高并发场景，有环境可以进行压力测试

 

6.为了查看更多线程信息，实现VisiableThreadPoolTaskExecutor  继承ThreadPoolTaskExecutor

package com.hbedu.search.hbsearch.utils;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import org.springframework.util.concurrent.ListenableFuture;



import java.util.concurrent.Callable;

import java.util.concurrent.Future;

import java.util.concurrent.ThreadPoolExecutor;


public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {


    private static final Logger logger = LoggerFactory.getLogger(VisiableThreadPoolTaskExecutor.class);


    private void showThreadPoolInfo(String prefix){

        ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();


        if(null==threadPoolExecutor){

            return;

        }


        logger.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");

        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);

    }

}

 

 

备注：项目为个人项目module ，单独执行时请注意