1.概述

2.springboot实现多线程

1.在入口类上开启多线程

@SpringBootApplication
//@EnableScheduling
@EnableAsync
public class SampleController {
    public static void main(String[] args) throws Exception {
        SpringApplication.run(SampleController.class, args);
    }
}

2.添加配置类

其实也可以直接在启动类中配置，但通常不会这么做

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
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 AsyncTaskConfig implements AsyncConfigurer {

    /**
     * 核心线程数
     */
//    @Value("${threadPool.corePoolSize}")
    private int corePoolSize = 10;
    /**
     * 最大线程数
     */
    private int maxPoolSize = 50;
    /**
     * 线程池缓冲队列容量
     */
    private int queueCapacity = 10;
    /**
     * 空闲线程销毁前等待时长
     */
    private int awaitTerminationSeconds = 10;
    /**
     * 线程名前缀
     */
    private String threadNamePrefix = "Sample-Async-";

    /**
     * ThreadPoolTaskExcutor运行原理
     * 当线程池的线程数小于corePoolSize，则新建线程入池处理请求
     * 当线程池的线程数等于corePoolSize，则将任务放入Queue中，线程池中的空闲线程会从Queue中获取任务并处理
     * 当Queue中的任务数超过queueCapacity，则新建线程入池处理请求，但如果线程池线程数达到maxPoolSize，将会通过RejectedExecutionHandler做拒绝处理
     * 当线程池的线程数大于corePoolSize时，空闲线程会等待keepAliveTime长时间，如果无请求可处理就自行销毁
     */
    @Override
    @Bean
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
        threadPool.setCorePoolSize(corePoolSize);
        threadPool.setMaxPoolSize(maxPoolSize);
        threadPool.setQueueCapacity(queueCapacity);
        threadPool.setAwaitTerminationSeconds(awaitTerminationSeconds);
        threadPool.setThreadNamePrefix(threadNamePrefix);

        //关机时，是否等待任务执行完
        threadPool.setWaitForTasksToCompleteOnShutdown(true);
        //设置拒绝策略
        //CALLER_RUNS：由调用者所在的线程执行该任务
        threadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        //初始化线程
        threadPool.initialize();
        return threadPool;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return null;
    }
}

3.编写任务类

import org.springframework.context.annotation.Scope;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

@Service
@Scope("prototype")
public class AsyncTaskTestService {

    @Async
    public void asyncTaskTest() {
        for (int i = 0; i < 100; i++) {
            System.out.println(Thread.currentThread().getName() + "aaaaaaaaaaaa");
        }
    }
}

 4.测试

我这里集成了swagger，就直接用controller测试了

    @Autowired
    private AsyncTaskTestService asyncTaskTestService;
    
    @PostMapping("/asyncTest")
    @ApiOperation(value = "多线程测试")
    public void asyncTest() {
        for (int i = 0; i < 5; i++) {
            asyncTaskTestService.asyncTaskTest();
        }
    }

 5.多线程使用过程中可能遇到的问题

一、线程池创建四种方式

Java通过Executors提供四种线程池，分别为：
newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
newScheduledThreadPool 创建一个定时线程池，支持定时及周期性任务执行。
newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

二、有返回值的多线程

ExecutorService接口继承自Executor，Executor中的execute方法无返回值，ExecutorService接口中的方法有返回值。

三、计数器使用

CountDownLatch也是juc包中的一个类，类似倒计时计数器，创建对象时通过构造方法设置初始值，调用CountDownLatch对象的await()方法则处于等待状态，调用countDown()方法就将计数器减1，当计数到达0时，则所有等待者或单个等待者开始执行。

有了计数器就可以暂时将主线程阻塞，等异步的多线程全部执行完毕并返回结果后，再继续执行主线程。

四、线程安全问题

有了上面线程池跟计数器的基础，现在可以动手写一个多线程处理任务并合并数据的demo了。

大致思路就是：创建一个定长的线程池，长度为10，计数器初始值也设置为10。每执行一次，将计数器减一，并且将执行结果添加到list集合中，最终多线程全部执行完毕后，计数器停止等待 主线程继续往下执行，返回list。

public static List<String> getExecutorService() throws InterruptedException{
		System.out.println("开始执行多线程...");
		long startTime = System.currentTimeMillis();
		List<String> list = new ArrayList<>();//存放返回结果
		CountDownLatch countDownLatch = new CountDownLatch(10);
		ExecutorService executorService = Executors.newFixedThreadPool(10);
		for (int i = 0; i < 10; i++) {
			Runnable runnable = new Runnable(){

				@Override
				public void run() {
					 try {
						Thread.sleep(3000);
                        list.add(UUID.randomUUID().toString());
                        System.out.println("当前线程name : "+Thread.currentThread().getName());
                        countDownLatch.countDown();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				
			};
			executorService.execute(runnable);
		}
		countDownLatch.await();
		System.out.println("submit总共cost 时间：" + (System.currentTimeMillis()-startTime)/1000 + "秒");
		executorService.shutdown();
		return list;
	}

 执行结果如下：

十个线程全部工作，但是返回值中却有null值，跟想要的结果有点出入，为啥呢？

原因在于：ArrayList是非线程安全的。ArrayList的add方法中有size++,不是一个原子操作，所以线程不安全。

 五、CopyOnWriteArrayList的用法

part4中提到的问题 解决方案很简单，将ArrayList换成CopyOnWriteArrayList即可。

	public static List<String> getExecutorService() throws InterruptedException{
		System.out.println("开始执行多线程...");
		long startTime = System.currentTimeMillis();
		List<String> list = new CopyOnWriteArrayList<>();//存放返回结果
		CountDownLatch countDownLatch = new CountDownLatch(10);
		ExecutorService executorService = Executors.newFixedThreadPool(10);
		for (int i = 0; i < 10; i++) {
			Runnable runnable = new Runnable(){

				@Override
				public void run() {
					 try {
						Thread.sleep(3000);
                        list.add(UUID.randomUUID().toString());
                        System.out.println("当前线程name : "+Thread.currentThread().getName());
                        countDownLatch.countDown();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				
			};
			executorService.execute(runnable);
		}
		countDownLatch.await();
		System.out.println("submit总共cost 时间：" + (System.currentTimeMillis()-startTime)/1000 + "秒");
		executorService.shutdown();
		return list;
	}

 CopyOnWriteArrayList是Java并发包中提供的一个并发容器，它是个线程安全且读操作无锁的ArrayList，写操作则通过创建底层数组的新副本来实现，是一种读写分离的并发策略，我们也可以称这种容器为"写时复制器"。

优点：读操作时性能很高，因为不需要任何同步措施，适用于读多写少的并发场景。

缺点：①.每次写操作都要copy原容器，频繁的GC，内存压力大。②.由于读写分离的策略，读到的数据很可能是旧数据。

java多线程执行任务

在项目开发的过程中经常会碰到多线程执行任务，每次用线程池实现时，由于每次的需求都有所差别有时是所有任务同时执行有时是分批次执行有时还需要知道所有任务什么时候执行完。今天闲着写了一个通用的多线程执行工具。

java多线程执行任务（工具升级版）

实现原理如图：

直接上代码

Scheder

import cn.hutool.core.util.RandomUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;


/**
 * @Author: dinghao
 * @Date: 2022/3/7 15:29
 */
@Slf4j
public class Scheder {

    /**
     * 任务信息数组
     */
    private TaskInfo[] taskInfos;

    /**
     * 执行计划队列
     */
    private LinkedBlockingQueue<Plan> planQueue;


    /**
     * 允许并行执行的线程数
     */
    private ExecutorService loopExecutor;

    /**
     * 执行模式：
     * 1：所有任务信息都执行
     * 2：先执行部分任务，执行完后再执行其他任务
     */
    private int model;

    /**
     * 先执行的数量
     */
    private int modelSize;

    /**
     * model = 2 时有效
     */
    private ArrayList<Integer> indexList = new ArrayList<>();

    /**
     * 协助判断，是否线程池的任务全部结束
     */
    private AtomicInteger count;

    /**
     * 调度器的执行状态
     */
    private volatile boolean status = false;


    /** 构造器
     * @param taskInfos 任务数组
     * @param nThrends  同时执行任务中的计划线程数
     * @param queueSize 计划执行队列
     * @param model     执行模式 1：所有任务信息都执行 2：先执行部分任务，执行完后再执行其他任务
     * @param modelSize 每批执行任务的数量
     */
    public Scheder(TaskInfo[] taskInfos, int nThrends, int queueSize, int model, Integer modelSize) {
        this.taskInfos = taskInfos;
        this.planQueue = new LinkedBlockingQueue<>(queueSize);
        this.loopExecutor = Executors.newFixedThreadPool(nThrends);
        this.model = model;
        if (this.model == 2 && modelSize != null) {
            this.modelSize = modelSize > taskInfos.length ? taskInfos.length : modelSize;
        } else {
            this.modelSize = taskInfos.length;
        }
		count = countPlan();
    }

    /**
     * 计算一共有多少执行计划
     * @return /
     */
    private AtomicInteger countPlan() {
        int sum = 0;
        for (int i = 0; i < this.taskInfos.length; i++) {
            sum += this.taskInfos[i].getPlanQueue().size();
        }
        return new AtomicInteger(sum);
    }

    public Scheder(TaskInfo[] taskInfos, int nThrends, int model, Integer modelSize) {
        this(taskInfos, nThrends, 100, model, modelSize);
    }

    public Scheder(TaskInfo[] taskInfos, int nThrends) {
        this(taskInfos, nThrends, 100, 1, null);
    }

    public Scheder(TaskInfo[] taskInfos) {
        this(taskInfos, 10, 100, 1, null);
    }

    public void setModel(int model) {
        this.model = model;
    }

    public int getModel() {
        return model;
    }

    public void setModelSize(int modelSize) {
        this.modelSize = modelSize;
    }

    public int getModelSize() {
        return modelSize;
    }

    public ArrayList<Integer> getIndexList() {
        return indexList;
    }


    public AtomicInteger getCount() {
        return count;
    }

    public boolean isStatus() {
        return status;
    }

    /**
     * 执行方法
     */
    public void run() {
        if (this.status) {
            log.warn("任务处于启动状态");
            return;
        }
        this.status = true;
        // 开启向队列中添加执行计划线程
        init();
        // 循环执行执行计划
        while (this.status) {
            // 所有执行计划执行完后，退出
            if (this.taskInfos.length <= 0 && this.planQueue.size() == 0) {
                this.status = false;
                break;
            }
            try {
                // 获取一个执行计划
                Plan plan = this.planQueue.take();
                // 执行计划
                this.loopExecutor.execute(() -> plan.run0(this.count));
            } catch (InterruptedException e) {
                log.error("任务执行中发生异常", e);
            }
        }
        int size;
        // 所有线程执行完毕出循环
        for (; ; ) {
            size = this.count.get();
            if (size == 0) {
                break;
            }
        }

        //停止线程池
        this.loopExecutor.shutdown();
        for (; ; ) {
            //只有当线程池中所有线程完成任务时才会返回true，并且需要先调用线程池的shutdown方法或者shutdownNow方法。
            if (this.loopExecutor.isTerminated()) {
                System.out.println("执行结束！");
                break;
            }
        }

    }

    /**
     * 开启一个线程，持续向执行计划队列添加执行计划，直到所有的计划任务添加完
     */
    private void init() {
        new Thread(() -> {
            while (this.status) {
                // 任务信息数组数量
                int length = this.taskInfos.length;
                // 执行完结束线程
                if (length <= 0) {
                    break;
                }
                // 获取添加执行计划的的任务索引值
                int index = getIndexOfModel(this.model, length);
                TaskInfo taskInfo = null;
                try {
                    taskInfo = this.taskInfos[index];

                } catch (Exception e) {
                    e.printStackTrace();
                }

                LinkedList<Plan> plans = taskInfo.getPlanQueue();
                if (plans.size() > 0) {
                    Plan plan = plans.removeFirst();
                    try {
                        this.planQueue.put(plan);
                    } catch (InterruptedException e) {
                        log.error("向执行计划队列放入计划异常", e);
                    }
                } else {
                    this.taskInfos = reBuildTaskInfos(this.taskInfos, index);
                }
            }
        }).start();
    }

    /**
     * 根据执行模式获取添加执行计划的的任务信息索引值
     *
     * @param model  执行模式
     * @param length 任务信息数组数量
     * @return 任务信息索引值
     */
    private int getIndexOfModel(int model, int length) {
        if (model == 1) {
            return RandomUtil.randomInt(0, length * 100) % length;
        } else {
        	this.indexList.removeIf(item -> item >= length);
            if (this.indexList.size() < this.modelSize) {
                int index = RandomUtil.randomInt(0, length * 100) % length;
                this.indexList.add(index);
                return index;
            } else {
                return this.indexList.get(RandomUtil.randomInt(0, length * 100) % this.indexList.size());
            }
        }
    }

    /**
     * 重新构建任务信息数组
     *
     * @param taskInfos 原来任务信息数组
     * @param index     需要移除的任务信息
     * @return 新的任务信息数组
     */
    private TaskInfo[] reBuildTaskInfos(TaskInfo[] taskInfos, int index) {
        TaskInfo[] newTaskINfo = new TaskInfo[taskInfos.length - 1];
        for (int j = 0, i = 0; i < taskInfos.length; i++) {
            if (i != index) {
                newTaskINfo[j] = taskInfos[i];
                j++;
            }
        }
        return newTaskINfo;
    }
}

构造器解释：

• @param taskInfos 任务数组
• @param nThrends 同时执行任务中的计划线程数
• @param queueSize 计划执行队列
• @param model 执行模式 1：所有任务信息都执行 2：先执行部分任务，执行完后再执行其他任务
• @param modelSize 每批执行任务的数量

TaskInfo

import lombok.Data;

import java.util.LinkedList;

/**
 * @Author: dinghao
 * @Date: 2022/3/7 15:31
 */
@Data
public class TaskInfo {

    /**
     * 任务名称
     */
    private String name;

    /**
     * 执行计划队列
     */
    private LinkedList<Plan> planQueue;


    public TaskInfo(String name, LinkedList<Plan> planQueue) {
        this.name = name;
        this.planQueue = planQueue;
    }
}

Plan

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: dinghao
 * @Date: 2022/3/7 15:37
 */
public interface Plan {

    /**
     * 线程池执行前
     */
    default void before(){
    }


    void run();


    /**
     * 线程池执行后
     */
    default void after(){
    }

    default void run0(AtomicInteger atomicInteger) {
        try{
            before();
            run();
        }finally {
            after();
            atomicInteger.decrementAndGet();
        }
    }


}

上面就是封装的工具

实现自己的计划

MyPlan

import lombok.Data;

/**
 * @Author: dinghao
 * @Date: 2022/3/9 10:33
 */
@Data
public class MyPlan implements Plan {
    private String name;

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + ":" + name);
    }
}

Test

public class Test {
	public static void main(String[] args) {
        int userSize = 1000;
        int jobSize = 100;

        TaskInfo[] taskInfos = new TaskInfo[userSize];


        IntStream.range(0, userSize).parallel().forEach(i -> {
            LinkedList<Plan> plans = new LinkedList<>();
            for (int j = 0; j < jobSize; j++) {
                MyPlan myPlan = new MyPlan();
                myPlan.setName("用户" + i + ",执行计划" + j);
                plans.add(myPlan);
            }
            taskInfos[i] = new TaskInfo("用户" + i, plans);
        });
        Scheder scheder = new Scheder(taskInfos, 10, 2, 10);
        scheder.run();
    }
}

测试结果：