Java并发编程艺术系列-四、 Java并发编程基础

四、 Java并发编程基础

本章将着重介绍Java并发编程的基础知识，从启动一个线程到线程间不同 的通信方式，最后通过简单的线程池示例以及应用（简单的Web服务器）来串联本章所介绍的内容。

4.1 线程

4.1.2 为什么要使用多线程

（1）利用更多的处理器核心

（2）更快的响应时间

（3）更好的编程模型

Java为多线程编程提供了良好、考究并且一致的编程模型，使开发人员能够更加专注于问题的解决，即为所遇到的问题建立合适的模型，而不是绞尽脑汁地考虑如何将其多线程化。一旦开发人员建立好了模型，稍做修改总是能够方便地映射到Java提供的多线程编程模型上。

4.1.3 线程优先级

• 在Java线程中，通过一个整型成员变量priority来控制优先级，优先级的范围从1~10，在线构建的时候可以通过setPriority(int)方法来修改优先级，

  默认优先级是5，优先级高的线程分配时间片的数量要多于优先级低的线程。

• 设置线程优先级时，针对频繁阻塞（休眠或者I/O操作）的线程需要设置较高优先级，

  而偏重计算（需要较多CPU时间或者偏运算）的线程则设置较低的优先级，确保处理器不会被独占。

• 注意 线程优先级不能作为程序正确性的依赖，因为操作系统可以完全不用理会Java线程对于优先级的设定

4.1.4 线程的状态

4.1.5 Daemon线程

• 当一个Java虚拟机中不存在非Daemon线程的时候，Java虚拟机将会退出。

• 通过调用Thread.setDaemon(true)将线程设置为Daemon线程。

• 注意 Daemon属性需要在启动线程之前设置，不能在启动线程之后设置。

public class Daemon {
    public static void main(String[] args) {
        Thread thread = new Thread(new DaemonRunner(), "DaemonRunner");
        thread.setDaemon(true);
        thread.start();
    }
    static class DaemonRunner implements Runnable {
        @Override
        public void run() {
            try {
                SleepUtils.second(10);
            } finally {
                System.out.println("DaemonThread finally run.");
            }
        }
    }
}

4.2 线程的启动与终止

4.2.1 线程构造

private void init(ThreadGroup g, Runnable target, String name, long stackSize, AccessControlContext acc) {
    if (name == null) {
        throw new NullPointerException("name cannot be null");
    }// 当前线程就是该线程的父线程 
    Thread parent = currentThread();
    this.group = g; // 将daemon、priority属性设置为父线程的对应属性 
    this.daemon = parent.isDaemon();
    this.priority = parent.getPriority();
    this.name = name.toCharArray();
    this.target = target;
    setPriority(priority); // 将父线程的InheritableThreadLocal复制过来 
    if (parent.inheritableThreadLocals != null)
        this.inheritableThreadLocals = ThreadLocal.createInheritedMap(parent.inheritableThreadLocals); // 分配一个线程ID 
    tid = nextThreadID();
}

4.2.2 启动线程

线程对象在初始化完成之后，调用start()方法就可以启动这个线程。线程start()方法的含义是：

当前线程（即parent线程）同步告知Java虚拟机，只要线程规划器空闲，应立即启动调用start()方法的线程。

• 注意 启动一个线程前，最好为这个线程设置线程名称，便于排查问题

4.2.3 理解中断

中断好比其他线程对该线程打了个招呼，其他线程通过调用该线程的interrupt() 方法对其进行中断操作

• 线程通过检查自身是否被中断来进行响应，通过方法isInterrupted()来进行判断是否被中断
• 也可以调用静态方法Thread.interrupted()对当前线程的中断标识位进行复位。
• 抛出InterruptedException的线程，其中断标识位会被清除

4.2.4 过期的suspend()、resume()和stop()

不建议使用的原因主要有：

以suspend()方法为例，在调用后，线程不会释放已经占有的资源（比如锁），而是占有着资源进入睡眠状态，这样容易引发死锁问题。

同样，stop()方法在终结一个线程时不会保证线程的资源正常释放，通常是没有给予线程完成资源释放工作的机会， 因此会导致程序可能工作在不确定状态下

正因为suspend()、resume()和stop()方法带来的副作用，这些方法才被标注为不建议使用的过期方法，而暂停和恢复操作可以用后面提到的等待/通知机制来替代。

4.2.5 安全地终止线程

使用条件变量安全终止线程实例

public class Shutdown {
    public static void main(String[] args) throws Exception {
        Runner one = new Runner();
        Thread countThread = new Thread(one, "CountThread");
        countThread.start(); // 睡眠1秒，main线程对CountThread进行中断，使CountThread能够感知中断而结束 
        TimeUnit.SECONDS.sleep(1);
        countThread.interrupt();
        Runner two = new Runner();
        countThread = new Thread(two, "CountThread");
        countThread.start(); // 睡眠1秒，main线程对Runner two进行取消，使CountThread能够感知on为false而结束 
        TimeUnit.SECONDS.sleep(1);
        two.cancel();
    }
    private static class Runner implements Runnable {
        private long i;
        private volatile boolean on = true;

        @Override
        public void run() {
            while (on && !Thread.currentThread().isInterrupted()) {
                i++;
            }
            System.out.println("Count i = " + i);
        }

        public void cancel() {
            on = false;
        }
    }
}

4.3 线程间通信

4.3.1 volatile和synchronized关键字

• 关键字volatile可以用来修饰字段（成员变量），就是告知程序任何对该变量的访问均需要从共享内存中获取，而对它的改变必须同步刷新回共享内存，

  它能保证所有线程对变量访问的可见性

• 关键字synchronized可以修饰方法或者以同步块的形式来进行使用，它主要确保多个线程在同一个时刻，只能有一个线程处于方法或者同步块中，

  它保证了线程对变量访问的可见性和排他性。

4.3.2 等待/通知机制

等待/通知机制，是指一个线程A调用了对象O的wait()方法进入等待状态，而另一个线程B调用了对象O的notify()或者notifyAll()方法，线程A收到通知后从对象O的wait()方法返回，进而执行后续操作。

下面例子中，创建了两个线程——WaitThread和NotifyThread，

前者检查flag值是否为false，如果符合要求，进行后续操作，否则在lock上等待，

后者在睡眠了一段时间后对lock进行通知，示例如下所示：

public class WaitNotify {
    static boolean flag = true;
    static Object lock = new Object();

    public static void main(String[] args) throws Exception {
        Thread waitThread = new Thread(new Wait(), "WaitThread");
        waitThread.start();
        TimeUnit.SECONDS.sleep(1);
        Thread notifyThread = new Thread(new Notify(), "NotifyThread");
        notifyThread.start();
    }
    static class Wait implements Runnable {
        public void run() { 
            // 加锁，拥有lock的Monitor 
            synchronized (lock) { 
                // 当条件不满足时，继续wait，同时释放了lock的锁 
                while (flag) {
                    try {
                        System.out.println(Thread.currentThread() + " flag is true. wait @ " + new SimpleDateFormat("HH:mm:ss").format(new Date()));
                        lock.wait();
                    } catch (InterruptedException e) {
                    }
                }
                // 条件满足时，完成工作
                System.out.println(Thread.currentThread() + " flag is false. running @ " + new SimpleDateFormat("HH:mm:ss").format(new Date()));
            }
        }
    }
    static class Notify implements Runnable {
        public void run() { 
            // 加锁，拥有lock的Monitor
            synchronized (lock) { 
                // 获取lock的锁，然后进行通知，通知时不会释放lock的锁，
                // 直到当前线程释放了lock后，WaitThread才能从wait方法中返回
                System.out.println(Thread.currentThread() + " hold lock. notify @ " + new SimpleDateFormat("HH:mm:ss").format(new Date()));
                lock.notifyAll();
                flag = false;
                SleepUtils.second(5);
            }
            // 再次加锁
            synchronized (lock) {
                System.out.println(Thread.currentThread() + " hold lock again. sleep @ " + new SimpleDateFormat("HH:mm:ss").format(new Date()));
                SleepUtils.second(5);
            }
        }
    }
}

4.3.3 等待/通知的经典范式

等待/通知的经典范式，该范式分为两部分，分别针对等待方（消费者）和通知方（生产者）。

• 等待方遵循如下原则。

  1）获取对象的锁。

  2）如果条件不满足，那么调用对象的wait()方法，被通知后仍要检查条件。

  3）条件满足则执行对应的逻辑。

  伪代码：

  synchronized(对象) { 
      while(条件不满足) { 
          对象.wait(); 
      }
      对应的处理逻辑 
  }
  

• 通知方遵循如下原则。

  1）获得对象的锁。

  2）改变条件。

  3）通知所有等待在对象上的线程。

  伪代码：

  synchronized(对象) { 
      改变条件 
      对象.notifyAll(); 
  }
  

4.3.4 管道输入/输出流

它主要用于线程之间的数据传输，而传输的媒介为内存。

管道输入/输出流主要包括了如下4种具体实现：PipedOutputStream、PipedInputStream、 PipedReader和PipedWriter，

前两种面向字节，而后两种面向字符。

在代码清单4-12所示的例子中，创建了printThread，它用来接受main线程的输入，任何main线程的输入均通过PipedWriter写入，而printThread在另一端通过PipedReader将内容读出并打印

public class Piped {
    public static void main(String[] args) throws Exception {
        PipedWriter out = new PipedWriter();
        PipedReader in = new PipedReader(); // 将输出流和输入流进行连接，否则在使用时会抛出IOException 
        out.connect(in);
        Thread printThread = new Thread(new Print(in), "PrintThread");
        printThread.start();
        int receive = 0;
        try {
            while ((receive = System.in.read()) != -1) {
                out.write(receive);
            }
        } finally {
            out.close();
        }
    }
    static class Print implements Runnable {
        private PipedReader in;

        public Print(PipedReader in) {
            this.in = in;
        }

        public void run() {
            int receive = 0;
            try {
                while ((receive = in.read()) != -1) {
                    System.out.print((char) receive);
                }
            } catch (IOException ex) {
            }
        }
    }
}

4.3.5 Thread.join()的使用

public class Join {
    public static void main(String[] args) throws Exception {
        Thread previous = Thread.currentThread();
        for (int i = 0; i < 10; i++) { // 每个线程拥有前一个线程的引用，需要等待前一个线程终止，才能从等待中返回 
            Thread thread = new Thread(new Domino(previous), String.valueOf(i));
            thread.start();
            previous = thread;
        }
        TimeUnit.SECONDS.sleep(5);
        System.out.println(Thread.currentThread().getName() + " terminate.");
    }

    static class Domino implements Runnable {
        private Thread thread;

        public Domino(Thread thread) {
            this.thread = thread;
        }

        public void run() {
            try {
                thread.join();
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread().getName() + " terminate.");
        }
    }
} 

• JDK中join源码

// 加锁当前线程对象 
public final synchronized void join() throws InterruptedException { 
    // 条件不满足，继续等待 
    while (isAlive()) { 
        wait(0); 
    }
    // 条件符合，方法返回 
}

4.3.6 ThreadLocal的使用

ThreadLocal，即线程变量，是一个以ThreadLocal对象为键、任意对象为值的存储结构。这个结构被附带在线程上，也就是说一个线程可以根据一个ThreadLocal对象查询到绑定在这个线程上的一个值。

通过set(T)方法来设置一个值，在当前线程下再通过get()方法获取到原先设置的值

public class Profiler { // 第一次get()方法调用时会进行初始化（如果set方法没有调用），每个线程会调用一次 
    private static final ThreadLocal<Long> TIME_THREADLOCAL = new ThreadLocal<Long>() {
        protected Long initialValue() {
            return System.currentTimeMillis();
        }
    };

    public static final void begin() {
        TIME_THREADLOCAL.set(System.currentTimeMillis());
    }

    public static final long end() {
        return System.currentTimeMillis() - TIME_THREADLOCAL.get();
    }

    public static void main(String[] args) throws Exception {
        Profiler.begin();
        TimeUnit.SECONDS.sleep(1);
        System.out.println("Cost: " + Profiler.end() + " mills");
    }
}

在AOP（面向方面编程）中，可以在方法调用前的切入点执行begin()方法，而在方法调用后的切入点执行 end()方法，这样依旧可以获得方法的执行耗时。

4.4 线程应用实例

4.4.1 等待超时模式

4.4.1.1 方法调用场景：

调用一个方法时等待一段时间（一般来说是给定一个时间段），如果该方法能够在给定的时间段之内得到结果，将结果立刻返回，否则超时返回默认结果。

4.4.1.2 对经典范式做出非常小的改动：

假设超时时间段是T，那么可以推断出在当前时间now+T之后就会超时。

定义如下变量。

·等待持续时间：REMAINING=T。

·超时时间：FUTURE=now+T。

这时仅需要wait(REMAINING)即可，在wait(REMAINING)返回之后会将执行： REMAINING=FUTURE–now。如果REMAINING小于等于0，表示已经超时，直接退出，否则将继续执行wait(REMAINING)。

// 对当前对象加锁 
public synchronized Object get(long mills) throws InterruptedException {
    long future = System.currentTimeMillis() + mills;
    long remaining = mills; // 当超时大于0并且result返回值不满足要求 
    while ((result == null) && remaining > 0) {
        wait(remaining);
        remaining = future - System.currentTimeMillis();
    }
    return result;
}

4.4.2 一个简单的数据库连接池示例

4.4.2.1 场景描述

我们使用等待超时模式来构造一个简单的数据库连接池，

在示例中模拟从连接池中获取、使用和释放连接的过程，而客户端获取连接的过程被设定为等待超时的模式，

也就是在1000毫秒内如果无法获取到可用连接，将会返回给客户端一个null。

设定连接池的大小为10个，然后通过调节客户端的线程数来模拟无法获取连接的场景。

4.4.2.2 分析设计

首先看一下连接池的定义。它通过构造函数初始化连接的最大上限，通过一个双向队列来维护连接，

调用方需要先调用fetchConnection(long)方法来指定在多少毫秒内超时获取连接，

当连接使用完成后，需要调用releaseConnection(Connection)方法将连接放回线程池，

示例如代码清单：

ConnectionPool.java

public class ConnectionPool {
    private LinkedList<Connection> pool = new LinkedList<Connection>();

    public ConnectionPool(int initialSize) {
        if (initialSize > 0) {
            for (int i = 0; i < initialSize; i++) {
                pool.addLast(ConnectionDriver.createConnection());
            }
        }
    }

    public void releaseConnection(Connection connection) {
        if (connection != null) {
            synchronized (pool) { 
                // 连接释放后需要进行通知，这样其他消费者能够感知到连接池中已经归还了一个连接 
                pool.addLast(connection);
                pool.notifyAll();
            }
        }
    }
    
    // 在mills内无法获取到连接，将会返回null 
    public Connection fetchConnection(long mills) throws InterruptedException {
        synchronized (pool) { 
            // 完全超时 
            if (mills <= 0) {
                while (pool.isEmpty()) {
                    pool.wait();
                }
                return pool.removeFirst();
            } else {
                long future = System.currentTimeMillis() + mills;
                long remaining = mills;
                while (pool.isEmpty() && remaining > 0) {
                    pool.wait(remaining);
                    remaining = future - System.currentTimeMillis();
                }
                Connection result = null;
                if (!pool.isEmpty()) {
                    result = pool.removeFirst();
                }
                return result;
            }
        }
    }
}

ConnectionDriver.java

通过动态代理构造了一个Connection，该Connection的代理实现仅仅是在commit()方法调用时休眠100毫秒，

public class ConnectionDriver {
    static class ConnectionHandler implements InvocationHandler {
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            if (method.getName().equals("commit")) {
                TimeUnit.MILLISECONDS.sleep(100);
            }
            return null;
        }
    }
    // 创建一个Connection的代理，在commit时休眠100毫秒 
    public static final Connection createConnection() {
        return (Connection) Proxy.newProxyInstance(ConnectionDriver.class.getClassLoader(), new Class<>[]{Connection.class}, new ConnectionHandler());
    }
}

• 测试代码ConnectionPoolTest.java

public class ConnectionPoolTest {
    static ConnectionPool pool = new ConnectionPool(10);
    // 保证所有ConnectionRunner能够同时开始
    static CountDownLatch start = new CountDownLatch(1);
    // main线程将会等待所有ConnectionRunner结束后才能继续执行
    static CountDownLatch end;

    public static void main(String[] args) throws Exception {
        // 线程数量，可以修改线程数量进行观察
        int threadCount = 10;
        end = new CountDownLatch(threadCount);
        int count = 20;
        AtomicInteger got = new AtomicInteger();
        AtomicInteger notGot = new AtomicInteger();
        for (int i = 0; i < threadCount; i++) {
            Thread thread = new Thread(new ConnetionRunner(count, got, notGot), "ConnectionRunnerThread");
            thread.start();
        }
        start.countDown();
        end.await();
        System.out.println("total invoke: " + (threadCount * count));
        System.out.println("got connection: " + got);
        System.out.println("not got connection " + notGot);
    }

    static class ConnetionRunner implements Runnable {
        int count;
        AtomicInteger got;
        AtomicInteger notGot;

        public ConnetionRunner(int count, AtomicInteger got, AtomicInteger notGot) {
            this.count = count;
            this.got = got;
            this.notGot = notGot;
        }

        public void run() {
            try {
                start.await();
            } catch (Exception ex) {
            }
            while (count > 0) {
                try {
                    // 从线程池中获取连接，如果1000ms内无法获取到，将会返回null
                    // 分别统计连接获取的数量got和未获取到的数量notGot
                    Connection connection = pool.fetchConnection(1000);
                    if (connection != null) {
                        try {
                            connection.createStatement();
                            connection.commit();
                        } finally {
                            pool.releaseConnection(connection);
                            got.incrementAndGet();
                        }
                    } else {
                        notGot.incrementAndGet();
                    }
                } catch (Exception ex) {
                } finally {
                    count--;
                }
            }
            end.countDown();
        }
    }
}

使用了CountDownLatch来确保ConnectionRunnerThread能够同时开始执行，并且在全部结束之后，才使main线程从等待状态中返回。

当前设定的场景是10个线程同时运行获取连接池（10个连接）中的连接，通过调节线程数量来观察未获取到连接的情况。

线程数、总获取次数、获取到的数量、未获取到的数量以及未获取到的比率

可以看出，在资源一定的情况下（连接池中的10个连接），随着客户端 线程的逐步增加，客户端出现超时无法获取连接的比率不断升高。

虽然客户端线程在这种超时获取的模式下会出现连接无法获取的情况，

但是它能够保证客户端线程不会一直挂在连接获取的操作上，而是“按时”返回，并告知客户端连接获取出现问题，是系统的一种自我保护机

制。数据库连接池的设计也可以复用到其他的资源获取的场景，针对昂贵资源（比如数据库连接）的获取都应该加以超时限制。

4.4.3 线程池技术及其示例

线程池技术预先创建了若干数量的线程，并且不能由用户直接对线程的创建进行控制，在这个前提下重复使用固定或较为固定数目的线程来完成任务的执行。

• 这样做的好处是，一方面，消除了频繁创建和消亡线程的系统资源开销，另一方面， 面对过量任务的提交能够平缓的劣化。

4.4.3.1 接口定义

ThreadPool.java

public interface ThreadPool<Job extends Runnable> { 
    // 执行一个Job，这个Job需要实现Runnable 
    void execute(Job job); 
    // 关闭线程池 
    void shutdown(); 
    // 增加工作者线程 
    void addWorkers(int num); 
    // 减少工作者线程 
    void removeWorker(int num); 
    // 得到正在等待执行的任务数量 
    int getJobSize(); 
}

4.4.3.2 默认实现DefaultThreadPool.java

public class DefaultThreadPool<Job extends Runnable> implements ThreadPool<Job> {
    // 线程池最大限制数 
    private static final int MAX_WORKER_NUMBERS = 10;
    // 线程池默认的数量 
    private static final int DEFAULT_WORKER_NUMBERS = 5;
    // 线程池最小的数量 
    private static final int MIN_WORKER_NUMBERS = 1;
    // 这是一个工作列表，将会向里面插入工作 
    private final LinkedList<Job> jobs = new LinkedList<Job>();
    // 工作者列表 
    private final List<Worker> workers = Collections.synchronizedList(new ArrayList<Worker>());
    // 工作者线程的数量 
    private int workerNum = DEFAULT_WORKER_NUMBERS;
    // 线程编号生成 
    private AtomicLong threadNum = new AtomicLong();

    public DefaultThreadPool() {
        initializeWokers(DEFAULT_WORKER_NUMBERS);
    }

    public DefaultThreadPool(int num) {
        workerNum = num > MAX_WORKER_NUMBERS MAX_WORKER_NUMBERS:
        num<MIN_WORKER_ NUMBERS MIN_WORKER_NUMBERS:
        num;
        initializeWokers(workerNum);
    }

    public void execute(Job job) {
        if (job != null) {
            // 添加一个工作，然后进行通知 
            synchronized (jobs) {
                jobs.addLast(job);
                jobs.notify();
            }
        }
    }

    public void shutdown() {
        for (Worker worker : workers) {
            worker.shutdown();
        }
    }

    public void addWorkers(int num) {
        synchronized (jobs) {
            // 限制新增的Worker数量不能超过最大值 
            if (num + this.workerNum > MAX_WORKER_NUMBERS) {
                num = MAX_WORKER_NUMBERS - this.workerNum;
            }
            initializeWokers(num);
            this.workerNum += num;
        }
    }

    public void removeWorker(int num) {
        synchronized (jobs) {
            if (num >= this.workerNum) {
                throw new IllegalArgumentException("beyond workNum");
            }
            // 按照给定的数量停止Worker 
            int count = 0;
            while (count < num) {
                Worker worker = workers.get(count) if (workers.remove(worker)) {
                    worker.shutdown();
                    count++;
                }
            } this.workerNum -= count;
        }
    }

    public int getJobSize() {
        return jobs.size();
    }

    // 初始化线程工作者 
    private void initializeWokers(int num) {
        for (int i = 0; i < num; i++) {
            Worker worker = new Worker();
            workers.add(worker);
            Thread thread = new Thread(worker, "ThreadPool-Worker-" + threadNum.incrementAndGet());
            thread.start();
        }
    }

    // 工作者，负责消费任务 
    class Worker implements Runnable {
        // 是否工作 
        private volatile boolean running = true;

        public void run() {
            while (running) {
                Job job = null;
                synchronized (jobs) {
                    // 如果工作者列表是空的，那么就wait 
                    while (jobs.isEmpty()) {
                        try {
                            jobs.wait();
                        } catch (InterruptedException ex) {
                            // 感知到外部对WorkerThread的中断操作，返回 
                            Thread.currentThread().interrupt();
                            return;
                        }
                    }
                    // 取出一个Job 
                    job = jobs.removeFirst();
                }
                if (job != null) {
                    try {
                        job.run();
                    } catch (Exception ex) { // 忽略Job执行中的Exception
                    }
                }
            }
        }

        public void shutdown() {
            running = false;
        }
    }
}

从线程池的实现可以看到，

• 当客户端调用execute(Job)方法时，会不断地向任务列表jobs中添加Job，而每个工作者线程会不断地从jobs上取出一个Job进行执行，当jobs为空时，工作者线程进入等待状态。

• 添加一个Job后，对工作队列jobs调用了其notify()方法，而不是notifyAll()方法，因为能够确定有工作者线程被唤醒，这时使用notify()方法将会比notifyAll()方法获得更小的开销（避免将等待队列中的线程全部移动到阻塞队列中）。

可以看到，线程池的本质就是使用了一个线程安全的工作队列连接工作者线程和客户端线程，

• 客户端线程将任务放入工作队列后便返回，

• 而工作者线程则不断地从工作队列上取出工作并执行。

• 当工作队列为空时，所有的工作者线程均等待在工作队列上，

• 当有客户端提交了一个任务之后会通知任意一个工作者线程，随着大量的任务被提交，更多的工作者线程会被唤醒。

4.4.4 注意

线程池中线程数量并不是越多越好，具体的数量需要评估每个任务的处理时间，以及当前计算机的处理器能力和数量。

使用的线程过少，无法发挥处理器的性能；使用的线程过多，将会增加系统的无故开销，起到相反的作用。