线程串行并行调度实现

线程串行并行调度实现

问题描述

问题描述：线程A、B、C并行执行，然后和线程D串行执行，如何实现。

问题具体化：现在有A、B、C三个线程，每一个线程分别完成打印0～4的任务，有一个线程D，在A、B、C完成打印之后打印“Hello world!”。

解决思路

我们都知道线程就是用来实现并发执行的，而要实现的结果是在A、B、C三个线程并行执行完之后才执行线程D的任务，那么就需要获取等到A、B、C三个线程的结果，之后再调用线程D的执行方法，就需要想办法让线程D暂不执行，等待A、B、C三个线程执行完才执行，根据这个思路，我们可以想到的解决方案方法如下：

• 通过一些flag变量控制判断A、B、C三个线程是否执行完

这是一个理解起来最简单也是最原始的方法，通过不断循环判断A、B、C三个线程对应的标识符是否完成执行来决定是否该执行线程D。

• 在线程D执行前阻塞线程，等待A、B、C三个线程执行完在执行

这个思路和上面差不多，只不过一个是循环判断，一个是阻塞线程。阻塞线程的方法有很多，sleep了，上锁了之类的，但是我们除了阻塞线程外，还需要有一个时机去唤醒阻塞，这个时机的触发点就是A、B、C三个线程执行完，综合考虑，能满足这样条件的阻塞方式有以下几种：

1、thread.join() 阻塞等待当前线程执行完

2、利用FeatureTask实现的线程，通过get（）方法可以阻塞等待线程结果返回

3、CountDownLatch，通过闭锁计数器的方式，通过await方法阻塞线程，在A、B、C三个线程执行完之后减少计数，唤醒阻塞

4、CyclicBarrier，栅栏原理，通过其await方法阻塞线程，在A、B、C三个线程执行完之后触发到达栅栏数，唤醒阻塞

以上就是能想到的实现方式，接下来一个一个用代码的方式实现并简要介绍其原理和特性。

解决方案及原理分析

通过flag变量控制

• 代码实现

//线程类
class MyRunnable implements Runnable {
    Flag flag;//变量
    CountDownLatch countDownLatch;//闭锁
    CyclicBarrier cyclicBarrier;//栅栏

    /**
     * 变量控制的方法
     *
     * @param flag
     */
    public MyRunnable(Flag flag) {
        this.flag = flag;
    }

    /**
     * 通过其他阻塞的方法时调用
     */
    public MyRunnable() {

    }
    /**
     * 通过CountDownLatch的方式
     */
    public MyRunnable(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }
    /**
     * 通过CyclicBarrier的方式
     */
    public MyRunnable(CyclicBarrier cyclicBarrier) {
        this.cyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        for (int i = 0; i < 5; i++) {
            try {
                Thread.sleep(new Random().nextInt(100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("当前线程：" + Thread.currentThread().getName() + " >> " + i);
        }
        if (countDownLatch != null) {//通过闭锁方式
            countDownLatch.countDown();
        }
        if (cyclicBarrier != null) {//通过栅栏方式
            try {
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
        if (flag != null) {//通过flag变量方式
            flag.setFlag(true);
            System.out.println("当前线程：" + Thread.currentThread().getName() + " >> " + flag.isFlag());
        }
    }
}

class Flag {
    boolean flag = false;

    public void setFlag(boolean flag) {
        this.flag = flag;
    }

    public boolean isFlag() {
        return flag;
    }
}

/**
 * 线程调度方法1 flag变量控制
 * Created by anonyper on 2019/10/17.
 */
public class ThreadTest {
    Flag flagA = new Flag();//考虑一下这里为什么用一个对象包装一个boolean而不是直接用boolean对象来传递呢
    Flag flagB = new Flag();
    Flag flagC = new Flag();

    public void testThread() {
        Thread threadA = new Thread(new MyRunnable(flagA), "A");
        Thread threadB = new Thread(new MyRunnable(flagB), "B");
        Thread threadC = new Thread(new MyRunnable(flagC), "C");
        Thread threadD = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("当前线程：" + Thread.currentThread().getName() + " >> Hello World!");
            }
        }, "D");
        threadA.start();
        threadB.start();
        threadC.start();
        while (!flagA.isFlag() || !flagB.isFlag() || !flagC.isFlag()) {
            Thread.yield();//释放CPU资源
            //this.wait(0L); //这样也可以等待 释放CPU资源
            //sleep(0L);//这样也可以等待 释放CPU资源
        }
        threadD.start();
    }

    public static void main(String[] args) {
        new ThreadTest().testThread();
    }
}

//执行结果：
当前线程：A >> 0
当前线程：B >> 0
当前线程：C >> 0
当前线程：A >> 1
当前线程：A >> 2
当前线程：B >> 1
当前线程：C >> 1
当前线程：B >> 2
当前线程：A >> 3
当前线程：B >> 3
当前线程：A >> 4
当前线程：A >> true
当前线程：B >> 4
当前线程：B >> true
当前线程：C >> 2
当前线程：C >> 3
当前线程：C >> 4
当前线程：C >> true
当前线程：D >> Hello World!

• 原理分析

通过wihle循环以及变量控制，让当前线程等待A、B、C三个线程执行完之后在执行D线程。这种方法是理解简单，但是实现挺麻烦的，线程越多要控制的变量就越多，非常不便。

注意实现Runnable中参数传递，这里涉及到值传递和对象传递的知识点。

Thread.join方法

• 代码实现

/**
 * 线程调度方法1 flag变量控制
 * Created by anonyper on 2019/10/17.
 */
public class ThreadTest {

    public void testThread() {
        Thread threadA = new Thread(new MyRunnable(), "A");
        Thread threadB = new Thread(new MyRunnable(), "B");
        Thread threadC = new Thread(new MyRunnable(), "C");
        Thread threadD = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("当前线程：" + Thread.currentThread().getName() + " >> Hello World!");
            }
        }, "D");
        threadA.start();
        threadB.start();
        threadC.start();
        try {
            threadA.join();//阻塞等待线程A运行完
            threadB.join();//阻塞等待线程B运行完
            threadC.join();//阻塞等待线程C运行完
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        threadD.start();
    }

    public static void main(String[] args) {
        new ThreadTest().testThread();
    }
}

//执行结果
当前线程：A >> 0
当前线程：C >> 0
当前线程：B >> 0
当前线程：B >> 1
当前线程：A >> 1
当前线程：C >> 1
当前线程：B >> 2
当前线程：C >> 2
当前线程：A >> 2
当前线程：C >> 3
当前线程：B >> 3
当前线程：C >> 4
当前线程：A >> 3
当前线程：A >> 4
当前线程：B >> 4
当前线程：D >> Hello World!

• 原理分析

线程的join方法会阻塞等待当前线程执行完成，其源代码如下：

public final void join() throws InterruptedException {
    this.join(0L);
}
public final synchronized void join(long var1) throws InterruptedException {
        long var3 = System.currentTimeMillis();//当前时间
        long var5 = 0L;
        if (var1 < 0L) {//等待时间不能小于0
            throw new IllegalArgumentException("timeout value is negative");
        } else {
            if (var1 == 0L) {//如果为零，则等待执行完成
                while(this.isAlive()) {//判断线程是否激活，涉及到线程状态，执行完之后返回false
                    this.wait(0L);//等待0秒后唤醒
                }
            } else {//等待一定时间后判断是否执行完
                while(this.isAlive()) {
                    long var7 = var1 - var5;
                    if (var7 <= 0L) {
                        break;
                    }

                    this.wait(var7);
                    var5 = System.currentTimeMillis() - var3;
                }
            }

        }
    }

我们看到join方法如果不传时间则默认等待线程执行完，但是其过程可能会被Interrupted，所以使用该方法需要针对异常做好判断。

FeatureTask的get方法

• 代码实现

/**
 * 实现Callable接口，重写call方法
 */
class MyCall implements Callable<Boolean> {
    @Override
    public Boolean call() {
        for (int i = 0; i < 5; i++) {
            try {
                Thread.sleep(new Random().nextInt(100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("当前线程：" + Thread.currentThread().getName() + " >> " + i);
        }
        return true;
    }
}

/**
 * 线程调度方法1 flag变量控制
 * Created by anonyper on 2019/10/17.
 */
public class ThreadTest {

    public void testThread() {
        FutureTask futureTaskA = new FutureTask(new MyCall());
        FutureTask futureTaskB = new FutureTask(new MyCall());
        FutureTask futureTaskC = new FutureTask(new MyCall());
        Thread threadA = new Thread(futureTaskA, "A");
        Thread threadB = new Thread(futureTaskB, "B");
        Thread threadC = new Thread(futureTaskC, "C");
        Thread threadD = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("当前线程：" + Thread.currentThread().getName() + " >> Hello World!");
            }
        }, "D");
        threadA.start();
        threadB.start();
        threadC.start();
        try {
            futureTaskA.get();
            futureTaskB.get();
            futureTaskC.get();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        threadD.start();
    }

    public static void main(String[] args) {
        new ThreadTest().testThread();
    }
}

//运行结果
当前线程：A >> 0
当前线程：C >> 0
当前线程：B >> 0
当前线程：A >> 1
当前线程：A >> 2
当前线程：C >> 1
当前线程：A >> 3
当前线程：B >> 1
当前线程：C >> 2
当前线程：C >> 3
当前线程：B >> 2
当前线程：A >> 4
当前线程：B >> 3
当前线程：C >> 4
当前线程：B >> 4
当前线程：D >> Hello World!

• 原理分析

1、Callable接口

public interface Callable<V> {
    V call() throws Exception;
}

2、Future接口

public interface Future<V> {
    boolean cancel(boolean var1);//取消

    boolean isCancelled();//是否取消

    boolean isDone();//是否完成

    V get() throws InterruptedException, ExecutionException;//返回结果

    V get(long var1, TimeUnit var3) throws InterruptedException, ExecutionException, TimeoutException;//等待具体时间内返回结果
}

2、RunnableFuture接口实现了Runnable、Future接口

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

3、FutureTask实现了RunnableFuture接口，构造函数接受一个Callable对象

public class FutureTask<V> implements RunnableFuture<V> {
    private Callable<V> callable;
    public FutureTask(Callable<V> var1) {
        if (var1 == null) {
            throw new NullPointerException();
        } else {
            this.callable = var1;
            this.state = 0;
        }
    }
    public void run() {
      ...
      Callable var1 = this.callable;
      var2 = var1.call;
      ...
    }
        public V get() throws InterruptedException, ExecutionException {//get方法
        int var1 = this.state;
        if (var1 <= COMPLETING) {//状态未完成时等待完成
            var1 = this.awaitDone(false, 0L);
        }

        return this.report(var1);//完成后返回结果
    }
    private int awaitDone(boolean var1, long var2) throws InterruptedException {
        long var4 = var1 ? System.nanoTime() + var2 : 0L;
        FutureTask.WaitNode var6 = null;
        boolean var7 = false;

        while(!Thread.interrupted()) {
            int var8 = this.state;
            if (var8 > COMPLETING) {//完成或被打断，返回结果
                if (var6 != null) {
                    var6.thread = null;
                }
                return var8;
            }

            if (var8 == COMPLETING) {//快要完成了，等待一会
                Thread.yield();
            } else if (var6 == null) {
                var6 = new FutureTask.WaitNode();
            } else if (!var7) {
                var7 = UNSAFE.compareAndSwapObject(this, waitersOffset, var6.next = this.waiters, var6);
            } else if (var1) {//有等待时间限制
                var2 = var4 - System.nanoTime();
                if (var2 <= 0L) {
                    this.removeWaiter(var6);
                    return this.state;
                }

                LockSupport.parkNanos(this, var2);
            } else {
                LockSupport.park(this);//通过LockSupport.park方法阻塞
            }
        }
                //最后没有等待结果时被唤醒（LockSupport.uppark()）抛出打断异常
        this.removeWaiter(var6);
        throw new InterruptedException();
    }

4、FutureTask对象作为runnable实现类，传入Thread中，调用run方法调用的是callable的call方法，get方法在call方法没有返回时阻塞线程，等待结果返回。

该方法学会了理解起来也简单，但是相对写的类比较多，如果比较在意线程结果返回值来做判断条件的话，可以使用。

CountDownLatch实现

• 代码实现

//MyRunnable的代码
        /**
     * 通过CountDownLatch的方式
     */
    public MyRunnable(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }
        @Override
    public void run() {
        ...
        if (countDownLatch != null) {//完成后减1
            countDownLatch.countDown();
        }
        ...
    }

public class ThreadTest {

    public void testThread() {
        CountDownLatch countDownLatch = new CountDownLatch(3);//闭锁的数量 三个线程
        //MyRunnable的实现见第一种方式代码，此处不在重复
        Thread threadA = new Thread(new MyRunnable(countDownLatch), "A");
        Thread threadB = new Thread(new MyRunnable(countDownLatch), "B");
        Thread threadC = new Thread(new MyRunnable(countDownLatch), "C");
        Thread threadD = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("当前线程：" + Thread.currentThread().getName() + " >> Hello World!");
            }
        }, "D");
        threadA.start();
        threadB.start();
        threadC.start();
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        threadD.start();
    }

    public static void main(String[] args) {
        new ThreadTest().testThread();
    }
}

//执行结果
当前线程：A >> 0
当前线程：B >> 0
当前线程：C >> 0
当前线程：A >> 1
当前线程：B >> 1
当前线程：A >> 2
当前线程：A >> 3
当前线程：C >> 1
当前线程：B >> 2
当前线程：A >> 4
当前线程：C >> 2
当前线程：B >> 3
当前线程：C >> 3
当前线程：B >> 4
当前线程：C >> 4
当前线程：D >> Hello World!

• 原理分析

CountDownLatch的核心方法有三个：

1、构造方式传入一个计数数量

private final CountDownLatch.Sync sync;

public CountDownLatch(int var1) {
    if (var1 < 0) {
        throw new IllegalArgumentException("count < 0");
    } else {
        this.sync = new CountDownLatch.Sync(var1);
    }
}

2、countDown计数减1

public void countDown() {
    this.sync.releaseShared(1);
}

3、await阻塞等待，还有一个方法是await(long var1, TimeUnit var3)，等待具体的时间

public void await() throws InterruptedException {
    this.sync.acquireSharedInterruptibly(1);
}

public boolean await(long var1, TimeUnit var3) throws InterruptedException {
    return this.sync.tryAcquireSharedNanos(1, var3.toNanos(var1));
}

这三个方法都关联一个类：Sync，该类继承AbstractQueuedSynchronizer类，简称AQS，AQS是用来构建锁或者其他同步组件（信号量、事件等）的基础框架类，JDK中许多并发工具类的内部实现都依赖于AQS，如ReentrantLock, Semaphore, CountDownLatch等等，具体源码此处不做分析。

CountDownLatch简单理解就是await方法阻塞线程，在等待计数为0时唤醒等待，减少计数的方法可以是在不同的线程中，也可以在同一个线程中。使用也比较简单，理解也容易。

CyclicBarrier实现

• 代码实现

//MyRunnable中的代码
/**
  * 通过CyclicBarrier的方式
  */
public MyRunnable(CyclicBarrier cyclicBarrier) {
  this.cyclicBarrier = cyclicBarrier;
}


@Override
public void run() {
  ...
    if (cyclicBarrier != null) {//通过栅栏方式
    try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
    }
  ...
}

public class ThreadTest {

    public void testThread() {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(4);//这个地方是4，不是3
        Thread threadA = new Thread(new MyRunnable(cyclicBarrier), "A");
        Thread threadB = new Thread(new MyRunnable(cyclicBarrier), "B");
        Thread threadC = new Thread(new MyRunnable(cyclicBarrier), "C");
        Thread threadD = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("当前线程：" + Thread.currentThread().getName() + " >> Hello World!");
            }
        }, "D");
        threadA.start();
        threadB.start();
        threadC.start();
        try {
            cyclicBarrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        threadD.start();
    }

    public static void main(String[] args) {
        new ThreadTest().testThread();
    }
}

//执行结果
当前线程：B >> 0
当前线程：B >> 1
当前线程：C >> 0
当前线程：B >> 2
当前线程：C >> 1
当前线程：B >> 3
当前线程：B >> 4
当前线程：A >> 0
当前线程：C >> 2
当前线程：C >> 3
当前线程：A >> 1
当前线程：C >> 4
当前线程：A >> 2
当前线程：A >> 3
当前线程：A >> 4
当前线程：D >> Hello World!

• 原理分析

CyclicBarrier是通过ReentrantLock锁来实现的，具体源码就不分析了，其里面也有几个重要的方法：

1、构造方法

public CyclicBarrier(int var1){//传入等待条件数
  ...
}

2、等待方法，当wait方法调用次数达到设定的次数之后，统一唤醒所有等待地方。

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return this.dowait(false, 0L);
    } catch (TimeoutException var2) {
        throw new Error(var2);
    }
}

CyclicBarrier和CountDownLatch有点类似，但是也有点不一样，用在跑步比赛上的区别就是：

1、CountDownLatch 计数减到0之后，阻塞的地方就可以开始跑了（阻塞了一个地方）

2、CyclicBarrier 当有运动员没有准备好（调用await方法）时，其他的运动员都等着（await阻塞），只有都准备好了再开始跑。

3、CountDownLatch计数只能用一次，CyclicBarrier可以循环使用。

所以用CyclicBarrier的实现思路，相当于让A、B、C三个线程阻塞在最后一步，然后线程D就绪，A、B、C三个线程收个尾之后线程D开始运行，严格意义上并不是A、B、C三个线程和D串行，在理论上实现了串行。

总结

以上就是我能想到的实现A、B、C三个线程并发然后和线程D串行执行的几种方法，代码都以贴出。如果有其他方法，欢迎提出补充。