Java并发常见问题

1.如何停止一个线程

前面有文章讲过Thread并没有一个cancel()方法，如果忘了为什么，回去翻翻文章。无论是通过线程的interrupt或者自行在Runnable中设置一个线程安全的变量来控制线程运行，核心的点都在于需要通过协作机制来完成线程的运行控制。也就是说要在Runnable或者Callable的实现方法中加入检查方法，不停的检查是否需要被中断。
同时在支持超时传参的一些方法中，尽量传入超时控制参数。例如JDBC读取超时参数或者Socket超时参数。

2.如何实现一个生产者消费者模型

• 使用ArrayBlockingQueue。当然面试官肯定无法接受这个回答，接下来的问题肯定是追问ArrayBlockingQueue实现原理。这是一个写代码方便但回答面试题不方便的做法，啊，这就是残酷的现实。
• 使用Lock + Condition实现一个自己的ArrayBlockingQueue。
  首先来实现一个简单的Queue叫MyBlockingQueue：

public class MyBlockingQueue<E> {
    private Queue<E> queue;
    private int max = 16;

    public MyBlockingQueue(int max) {
        queue = new LinkedList<>();
        this.max = max;
    }

    public void put(E e) {
       queue.add(e);
    }

    public E take() {
        E remove = queue.remove();
        return remove;
    }
}

接下来需要加上同步访问控制功能，确保同一时间只能有一个线程访问Queue :

public class MyBlockingQueue<E> {
    private Queue<E> queue;
    private int max = 16;
    private ReentrantLock reentrantLock = new ReentrantLock(true);

    public MyBlockingQueue(int max) {
        queue = new LinkedList<>();
        this.max = max;
    }

    public void put(E e) {
        reentrantLock.lock();
        try {
            queue.add(e);
        } finally {
            reentrantLock.unlock();
        }
    }

    public E take() {
        reentrantLock.lock();
        try {
            E remove = queue.remove();
            return remove;
        } finally {
            reentrantLock.unlock();
        }

        return null;
    }
}

接下来要做的是根据Queue的元素个数控制访问线程的等待或者唤醒状态，对于put(E e)来说就是当Queue中的元素个数到达max个时就block住访问的线程，对于take()来说就是当Queue中的元素个数为0时就block住访问的线程，可以通过Condition来控制线程状态:

public class MyBlockingQueue<E> {
    private Queue<E> queue;
    private int max = 16;
    private ReentrantLock reentrantLock = new ReentrantLock(true);
    private Condition notEmpty = reentrantLock.newCondition();
    private Condition notFull = reentrantLock.newCondition();

    public MyBlockingQueue(int max) {
        queue = new LinkedList<>();
        this.max = max;
    }

    public void put(E e) {
        reentrantLock.lock();
        try {
            if (queue.size() == max) {
                notFull.await();//等待其他线程取走元素后唤醒
            }
            queue.add(e);
            notEmpty.signalAll();//queue中可以take元素唤醒
        }  finally {
            reentrantLock.unlock();
        }
    }

    public E take() {
        reentrantLock.lock();
        try {
            if (queue.size() == 0) {
                notEmpty.await();//等待其他线程添加元素后唤醒
            }
            E remove = queue.remove();
            notFull.signalAll();//queue中可以put元素唤醒
            return remove;
        } finally {
            reentrantLock.unlock();
        }

        return null;
    }
}

这样就算是初步完成了一个简单的BlockingQueue，但是这是有问题的。注意看下面修改过的代码:

package com.sync.exe;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class MyBlockingQueue<E> {
    private Queue<E> queue;
    private int max = 16;
    private ReentrantLock reentrantLock = new ReentrantLock(true);
    private Condition notEmpty = reentrantLock.newCondition();
    private Condition notFull = reentrantLock.newCondition();

    public MyBlockingQueue(int max) {
        queue = new LinkedList<>();
        this.max = max;
    }

    public void put(E e) {
        reentrantLock.lock();
        try {
            while (queue.size() == max) {//注意while和上面的if对比
                notFull.await();
            }
            queue.add(e);
            notEmpty.signalAll();
        } finally {
            reentrantLock.unlock();
        }
    }

    public E take() {
        reentrantLock.lock();
        try {
            while (queue.size() == 0) {//注意while和上面的if对比
                notEmpty.await();
            }
            E remove = queue.remove();
            notFull.signalAll();
            return remove;
        } finally {
            reentrantLock.unlock();
        }
    }
}

为什么要换成while？假设这样一个情况，某个同事（我说的这个同事到底是不是我自己？？？）在扩展该类的时候错误的调用了full.signalAll();或者notFull.signalAll();方法，此时恰好有一个线程卡在await()这里。如果使用if就会直接走下面的代码，那么就会超出max限制或者返回null，这是设计这个类时所不允许的。可以通过上面的陈述来感知一下两者的差异。

3.如何从不同的电商获取一个商品的价格，并在指定时间内返回（Timeout机制）？

先让我们来分析下这个功能如何去实现。大概可以拆解成两大步：

1. 使用多个线程同时去获取价格
2. 指定这些线程执行的timeout时间

具体实现：

• 使用CountDownLatch:

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        CountDownLatch countDownLatch = new CountDownLatch(3);
        executorService.execute(new FetchPrice(countDownLatch));
        executorService.execute(new FetchPrice(countDownLatch));
        executorService.execute(new FetchPrice(countDownLatch));


        try {
            countDownLatch.wait(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
}

    static class FetchPrice implements Runnable {

        private CountDownLatch countDownLatch;

        public FetchPrice(CountDownLatch countDownLatch) {
            countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            //fetch price
            countDownLatch.countDown();//注意点
        }
    }

不难看出，使用CountDownLatch的一些要点。首先是设置任务个数，然后是在每个任务完成后countDownLatch.countDown();，最后是使用countDownLatch.wait(5000)来增加超时机制。

• 使用CompletableFuture（since 1.8）：

    public static void main(String[] args) {
        CompletableFuture<Void> task1 = CompletableFuture.runAsync(new FetchPrice2());
        CompletableFuture<Void> task2 = CompletableFuture.runAsync(new FetchPrice2());
        CompletableFuture<Void> task3 = CompletableFuture.runAsync(new FetchPrice2());

        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(task1, task2, task3);
        try {
            voidCompletableFuture.get(5000, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    static class FetchPrice2 implements Runnable {

        @Override
        public void run() {
            //fetch price
        }
    }

同样很简单，具体这些类的api可以自行查看，和Executors有很多相似的地方。