Java和Springboot中的多线程

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

 /**
     * Causes this thread to begin execution; the Java Virtual Machine
     * calls the <code>run</code> method of this thread.
     * 
     * 1、start方法将导致this thread开始执行。由JVM调用this thread的run方法。
     * 
     * The result is that two threads are running concurrently: the
     * current thread (which returns from the call to the
     * <code>start</code> method) and the other thread (which executes its
     * <code>run</code> method).
     * 
     * 2、结果是 调用start方法的当前线程 和 执行run方法的另一个线程 并发运行。
     * 
     * It is never legal to start a thread more than once.
     * In particular, a thread may not be restarted once it has completed
     * execution.
     *
     * 3、多次启动线程永远不合法。 特别是，线程一旦完成执行就不会重新启动。
     * 
     * @exception  IllegalThreadStateException  if the thread was already started.
     * 如果线程已启动，则抛出异常。
     * @see        #run()
     * @see        #stop()
     */
    public synchronized void start() {
        /**
         * This method is not invoked for the main method thread or "system"
         * group threads created/set up by the VM. Any new functionality added
         * to this method in the future may have to also be added to the VM.
         * 
         * 4、对于由VM创建/设置的main方法线程或“system”组线程，不会调用此方法。 
         *    未来添加到此方法的任何新功能可能也必须添加到VM中。
         * 
         * A zero status value corresponds to state "NEW".
         * 5、status=0 代表是 status 是 "NEW"。
         */
        if (threadStatus != 0)
            throw new IllegalThreadStateException();

        /* Notify the group that this thread is about to be started
         * so that it can be added to the group's list of threads
         * and the group's unstarted count can be decremented. 
         * 
         * 6、通知组该线程即将启动，以便将其添加到线程组的列表中，
         *    并且减少线程组的未启动线程数递减。
         * 
         * */
        group.add(this);

        boolean started = false;
        try {
            //7、调用native方法，底层开启异步线程，并调用run方法。
            start0();
            started = true;
        } finally {
            try {
                if (!started) {
                    group.threadStartFailed(this);
                }
            } catch (Throwable ignore) {
                /* do nothing. If start0 threw a Throwable then it will be passed up the call stack 
                 * 8、忽略异常。 如果start0抛出一个Throwable，它将被传递给调用堆栈。
                 */
            }
        }
    }

 //native方法，JVM创建并启动线程，并调用run方法
 private native void start0();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

public class ThreadTest {
    public static void main(String[] args) {

        StopThread stopThread = new StopThread();
        // 启动线程
        stopThread.start();
        try {
            // 休眠1秒，确保线程进入运行
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 暂停线程
        stopThread.stop();
//        stopThread.interrupt();
        // 确保线程已经销毁
        while (stopThread.isAlive()) { }
        // 输出结果
        stopThread.print();

    }

    private static class StopThread extends Thread {

        private int x = 0;
        private int y = 0;

        @Override
        public void run(){
            // 这是一个同步原子操作
            synchronized (this){
                ++x;
                try {
                    // 休眠3秒,模拟耗时操作
                    Thread.sleep(3000);

                }catch (Exception e){
                    e.printStackTrace();
                }
                ++y;
            }
        }
        public void print() {
            System.out.println("x=" + x + " y=" + y);
        }



    }
}

1

x=1 y=0

1
2
3
4

java.lang.InterruptedException: sleep interrupted
	at java.lang.Thread.sleep(Native Method)
	at ThreadTest$StopThread.run(ThreadTest.java:35)
x=1 y=1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

   // 核心 interrupt 方法
   public void interrupt() {
        if (this != Thread.currentThread()) // 非本线程，需要检查权限
            checkAccess();

        synchronized (blockerLock) {
            Interruptible b = blocker;
            if (b != null) {
                interrupt0();           // 仅仅设置interrupt标志位
                b.interrupt(this);    // 调用如 I/O 操作定义的中断方法
                return;
            }
        }
        interrupt0();
    }
    // 静态方法，这个方法有点坑，调用该方法调用后会清除中断状态。
    public static boolean interrupted() {
        return currentThread().isInterrupted(true);
    }
    // 这个方法不会清除中断状态
    public boolean isInterrupted() {
        return isInterrupted(false);
    }
   // 上面两个方法会调用这个本地方法，参数代表是否清除中断状态
   private native boolean isInterrupted(boolean ClearInterrupted);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

public class ThreadTest {
    
    public static void main(String[] args) throws InterruptedException {
        Thread t = new Thread(new Task("1"));
        t.start();
        t.interrupt();
    }

    static class Task implements Runnable{
        String name;
        
        public Task(String name) {
            this.name = name;
        }
        
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println("thread has been interrupt!");
            }
            System.out.println("isInterrupted: " + Thread.currentThread().isInterrupted());
            System.out.println("task " + name + " is over");
        }
    }
}

1
2
3

thread has been interrupt!
isInterrupted: false
task 1 is over

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

public class ThreadTest {
    
    public static void main(String[] args) throws InterruptedException {
        Thread t = new Thread(new Task("1"));
        t.start();
        t.interrupt();
    }

    static class Task implements Runnable{
        String name;
        
        public Task(String name) {
            this.name = name;
        }
        
        @Override
        public void run() {
            System.out.println("first :" + Thread.interrupted());
            System.out.println("second:" + Thread.interrupted());
            System.out.println("task " + name + " is over");
        }
    }
}

1
2
3

first :true
second:false
task 1 is over

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

public class DemoThreadTask implements Runnable{
    @Override
    public void run() {
        // TODO Auto-generated method stub
    }
    
    public static void main(String[] args) {
        DemoThreadTask task = new DemoThreadTask();
        Thread t = new Thread(task);
        t.start();
        ...
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public class Thread implements Runnable {
    /* What will be run. */
    private Runnable target;

    public void run() {
        if (target != null) {
            target.run();
        }
   }
     ...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

public class DemoThread extends Thread{
    @Override 
    //重写run方法
    public void run() {
        // TODO Auto-generated method stub
    }

    public static void main(String[] args) {
        DemoThread t = new DemoThread();
        t.start();
        ...
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

public class DemoCallable implements Callable<String>{
    @Override
    public String call() throws Exception {
        // TODO Auto-generated method stub
        return null;
    }
    
    public static void main(String[] args) throws Exception {
        DemoCallable c = new DemoCallable();
        FutureTask<String> future = new FutureTask<>(c); 
        Thread t = new Thread(future);
        t.start();
        ...
        String result = future.get(); //同步获取返回结果
        System.out.println(result);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

public class Demo{
    
    public static void main(String[] args) throws Exception {
        //方式一：Thread匿名内部类
        new Thread(){
            @Override
            public void run() {
                // TODO Auto-generated method stub
            }
        }.start();
        
        //方式二：Runnable匿名内部类
        new Thread(new Runnable() {
            @Override
            public void run() {
                // TODO Auto-generated method stub
            }
        }).start();
        
        ...
    }
}

1
2
3
4
5
6

public class Demo{
    public static void main(String[] args) throws Exception {
        new Thread(() -> System.out.println("running") ).start() ;
        ...
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

public class DemoThreadTask implements Runnable{
    @Override
    public void run() {
        // TODO Auto-generated method stub
        System.out.println("running");
    }
    
    public static void main(String[] args) {
        DemoThreadTask task = new DemoThreadTask();
        ExecutorService ex = Executors.newCachedThreadPool();
        ex.execute(task);
        ...
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

public class DemoTimmerTask {

    public static void main(String[] args) throws Exception {
        Timer timer = new Timer();
        timer.scheduleAtFixedRate((new TimerTask() {
            @Override
            public void run() {
                System.out.println("定时任务1执行了....");
            }
        }), 2000, 1000);
    }
}

1

注：这里没有提到ScheduledExecutorService ，后续解析。

1

这个过程参考[线程任务处理流程图]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

 public ThreadPoolExecutor(int corePoolSize, // 1
                              int maximumPoolSize,  // 2
                              long keepAliveTime,  // 3
                              TimeUnit unit,  // 4
                              BlockingQueue<Runnable> workQueue, // 5
                              ThreadFactory threadFactory,  // 6
                              RejectedExecutionHandler handler ) { //7
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

1
2
3
4
5

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

1
2
3
4
5

     public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

1
2
3
4
5
6

    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

    public static void main(String[] args) {
        ExecutorService fixedExecutorService = Executors.newFixedThreadPool(1);
        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) fixedExecutorService;
        System.out.println(threadPoolExecutor.getMaximumPoolSize());
        threadPoolExecutor.setCorePoolSize(8);
        
        ExecutorService singleExecutorService = Executors.newSingleThreadExecutor();
//      运行时异常 java.lang.ClassCastException
//      ThreadPoolExecutor threadPoolExecutor2 = (ThreadPoolExecutor) singleExecutorService;
    }

1
2
3

    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

1
2
3
4

    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

import java.io.IOException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadPoolTest {

    public static void main(String[] args) {
        int corePoolSize = 2;
        int maximumPoolSize = 4;
        long keepAliveTime = 10;
        TimeUnit unit = TimeUnit.SECONDS;
//        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2);
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(2);
        ThreadFactory threadFactory = new NameThreadFactory();
        RejectedExecutionHandler rejectedExecutionHandler = new MyIgnorePolicy();
        ThreadPoolExecutor tpexecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, rejectedExecutionHandler);
        // 预启动所有核心线程
        tpexecutor.prestartAllCoreThreads();

        for (int i = 1; i <= 10; i++) {
            MyTask myTask = new MyTask(String.valueOf(i));
            tpexecutor.execute(myTask);
        }

        try {
            //阻塞主线程
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    static class NameThreadFactory implements ThreadFactory {

        private final AtomicInteger mThreadNum = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
            System.out.println(t.getName() + " has been created");
            return t;
        }
    }

    public static class MyIgnorePolicy implements RejectedExecutionHandler {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            doLog(r, executor);
        }

        private void doLog(Runnable r, ThreadPoolExecutor executor) {
            System.err.println(r.toString() + " rejected");
            System.out.println("getCompletedTaskCount: " + executor.getCompletedTaskCount());

        }
    }

    public static class MyTask implements Runnable {
        private String name;

        public MyTask(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            try {
                System.out.println(this.toString() + "is running!");
                //让任务执行慢点
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public String getName() {
            return name;
        }

        public String toString() {
            return "MyTask [name=" + name + "]";
        }

    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

    //原子状态控制数
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    //29比特位
    private static final int COUNT_BITS = Integer.SIZE - 3;
    //实际容量 2^29-1
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    // runState存储在高位中
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;

    // Packing and unpacking ctl 打包和解压ctl

    // 解压runState
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    // 解压workerCount
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    // 打包ctl
    private static int ctlOf(int rs, int wc) { return rs | wc; }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        // 1、工作线程 < 核心线程 
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 2、运行态，并尝试将任务加入队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        } // 3、使用尝试使用最大线程运行
        else if (!addWorker(command, false))
            reject(command);
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

private boolean addWorker(Runnable firstTask, boolean core) {
    // 第一部分：自旋、CAS、重读ctl 等结合，直到确定是否可以创建worker，
    // 可以则跳出循环继续操作，否则返回false
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c)) // CAS增长workerCount，成功则跳出循环
                break retry;
            c = ctl.get();  // Re-read ctl 重新获取ctl
            if (runStateOf(c) != rs) // 状态改变则继续外层循环，否则在内层循环
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
    // 第二部分：创建worker，这部分使用ReentrantLock锁
    boolean workerStarted = false; // 线程启动标志位
    boolean workerAdded = false;  // 线程是否加入workers 标志位
    Worker w = null;
    try {
        w = new Worker(firstTask); //创建worker
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 获取到锁以后仍需检查ctl，可能在上一个获取到锁处理的线程可能会改变runState
                // 如 ThreadFactory 创建失败 或线程池被 shut down等
                int rs = runStateOf(ctl.get());

                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive())
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start(); // 启动线程
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w); // 失败操作
    }
    return workerStarted;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

private final class Worker extends AbstractQueuedSynchronizer implements Runnable
{
    private static final long serialVersionUID = 6138294804551838833L;

    /** 每个worker有自己的内部线程，ThreadFactory创建失败时是null */
    final Thread thread;
    /** 初始化任务，可能是null */
    Runnable firstTask;
    /** 每个worker的完成任务数 */
    volatile long completedTasks;

    Worker(Runnable firstTask) {
        setState(-1); // 禁止线程在启动前被打断
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /** 重要的执行方法  */
    public void run() {
        runWorker(this);
    }

    // state = 0 代表未锁；state = 1 代表已锁

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }
    // interrupt已启动线程
    void interruptIfStarted() {
        Thread t;
        // 初始化是 state = -1，不会被interrupt
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // 允许被 interrupt
        boolean completedAbruptly = true;
        try {
            // loop 直至 task = null （线程池关闭、超时等）
            // 注意这里的getTask()方法，我们配置的阻塞队列会在这里起作用
            while (task != null || (task = getTask()) != null) {
                w.lock();  // 执行任务前上锁
                // 如果线程池停止，确保线程中断; 如果没有，确保线程不中断。这需要在第二种情况下进行重新获取ctl，以便在清除中断时处理shutdownNow竞争
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task); // 扩展点
                    Throwable thrown = null;
                    try {
                        task.run(); // 真正执行run方法
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown); // 扩展点
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly); // 线程退出工作
        }
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

private Runnable getTask() {
    boolean timedOut = false; // 上一次 poll() 是否超时

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 是否继续处理任务 可以参见上一篇的状态控制
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // 是否允许超时
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

@Configuration
@EnableAsync
public class AsyncConfiguration {

    @Bean("doSomethingExecutor")
    public Executor doSomethingExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数：线程池创建时候初始化的线程数
        executor.setCorePoolSize(10);
        // 最大线程数：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
        executor.setMaxPoolSize(20);
        // 缓冲队列：用来缓冲执行任务的队列
        executor.setQueueCapacity(500);
        // 允许线程的空闲时间60秒：当超过了核心线程之外的线程在空闲时间到达之后会被销毁
        executor.setKeepAliveSeconds(60);
        // 线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池
        executor.setThreadNamePrefix("do-something-");
        // 缓冲队列满了之后的拒绝策略：由调用线程处理（一般是主线程）
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
        executor.initialize();
        return executor;
    }

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

@RestController
public class AsyncController {

    @Autowired
    private AsyncService asyncService;

    @GetMapping("/open/something")
    public String something() {
        int count = 10;
        for (int i = 0; i < count; i++) {
            asyncService.doSomething("index = " + i);
        }
        lon
        return "success";
    }
}


@Slf4j
@Service
public class AsyncService {

    // 指定使用beanname为doSomethingExecutor的线程池
    @Async("doSomethingExecutor")
    public String doSomething(String message) {
        log.info("do something, message={}", message);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            log.error("do something error: ", e);
        }
        return message;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

2020-04-19 23:42:42.486  INFO 21168 --- [io-8200-exec-17] x.g.b.system.controller.AsyncController  : do something end, time 8 milliseconds
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-1] x.gits.boot.system.service.AsyncService  : do something, message=index = 0
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-5] x.gits.boot.system.service.AsyncService  : do something, message=index = 4
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-4] x.gits.boot.system.service.AsyncService  : do something, message=index = 3
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-6] x.gits.boot.system.service.AsyncService  : do something, message=index = 5
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-9] x.gits.boot.system.service.AsyncService  : do something, message=index = 8
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-8] x.gits.boot.system.service.AsyncService  : do something, message=index = 7
2020-04-19 23:42:42.488  INFO 21168 --- [do-something-10] x.gits.boot.system.service.AsyncService  : do something, message=index = 9
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-7] x.gits.boot.system.service.AsyncService  : do something, message=index = 6
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-2] x.gits.boot.system.service.AsyncService  : do something, message=index = 1
2020-04-19 23:42:42.488  INFO 21168 --- [ do-something-3] x.gits.boot.system.service.AsyncService  : do something, message=index = 2

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

@RestController
public class AsyncController {

    @Autowired
    private AsyncService asyncService;

    @SneakyThrows
    @ApiOperation("异步 有返回值")
    @GetMapping("/open/somethings")
    public String somethings() {
        CompletableFuture<String> createOrder = asyncService.doSomething1("create order");
        CompletableFuture<String> reduceAccount = asyncService.doSomething2("reduce account");
        CompletableFuture<String> saveLog = asyncService.doSomething3("save log");

        // 等待所有任务都执行完
        CompletableFuture.allOf(createOrder, reduceAccount, saveLog).join();
        // 获取每个任务的返回结果
        String result = createOrder.get() + reduceAccount.get() + saveLog.get();
        return result;
    }
}


@Slf4j
@Service
public class AsyncService {

    @Async("doSomethingExecutor")
    public CompletableFuture<String> doSomething1(String message) throws InterruptedException {
        log.info("do something1: {}", message);
        Thread.sleep(1000);
        return CompletableFuture.completedFuture("do something1: " + message);
    }

    @Async("doSomethingExecutor")
    public CompletableFuture<String> doSomething2(String message) throws InterruptedException {
        log.info("do something2: {}", message);
        Thread.sleep(1000);
        return CompletableFuture.completedFuture("; do something2: " + message);
    }

    @Async("doSomethingExecutor")
    public CompletableFuture<String> doSomething3(String message) throws InterruptedException {
        log.info("do something3: {}", message);
        Thread.sleep(1000);
        return CompletableFuture.completedFuture("; do something3: " + message);
    }
}

1
2

C:\Users\Administrator>curl -X GET "http://localhost:8200/open/somethings" -H "accept: */*"
do something1: create order; do something2: reduce account; do something3: save log

1
2
3

2020-04-20 00:27:42.238  INFO 5672 --- [ do-something-3] x.gits.boot.system.service.AsyncService  : do something3: save log
2020-04-20 00:27:42.238  INFO 5672 --- [ do-something-2] x.gits.boot.system.service.AsyncService  : do something2: reduce account
2020-04-20 00:27:42.238  INFO 5672 --- [ do-something-1] x.gits.boot.system.service.AsyncService  : do something1: create order