Java 并发编程的艺术（二）

并发编程的挑战

上下文切换

• 即使是单核处理器也支持多线程执行代码，CPU 通过给每个线程分配 CPU 时间片来实现这个机制。
• CPU 通过时间片分配算法来循环执行任务，当前任务执行一个时间片后会切换到下一个任务。
• 任务从保存到再加载的过程就是一次上下文切换
• 切换上下文需要一定的消耗

如何减少上下文的切换

• 使用最少线程：避免创建不需要的线程
• CAS 算法：Java 的 Atomic 包使用 CAS 算法来更新数据，而不需要加锁。
• 无锁并发编程：多线程竞争锁时，会引起上下文切换，所以多线程处理数据时，可以用一些办法来避免使用锁

死锁

• 避免一个线程同时获取多个锁。

• 避免一个线程在锁内同时占用多个资源，尽量保证每个锁只占用一个资源。

• 尝试使用定时锁，使用 lock.tryLock（timeout）来替代使用内部锁机制。

• 对于数据库锁，加锁和解锁必须在一个数据库连接里，否则会出现解锁失败的情况

资源限制的挑战

• 在并发编程中，将代码执行速度加快的原则是将代码中串行执行的部分变成并发执行，但是如果将某段串行的代码并发执行，因为受限于资源，仍然在串行执行，这时候程序不仅不会加快执行，反而会更慢，因为增加了上下文切换和资源调度的时间

Java 并发机制的底层实现原理

volatile 的应用

• volatile保证共享变量的可见性
• volatile 是轻量级的 synchronized

synchronized 的实现原理与应用

• 对于普通同步方法，锁是当前实例对象。

• 对于静态同步方法，锁是当前类的 Class 对象。

• 对于同步方法块，锁是 Synchonized 括号里配置的对象。

三大特性

• 原子性：一个或多个操作要么全部成功，要么全部失败。保证只有一个线程拿到锁访问共享资源
• 可见性：当一个线程对共享变量进行修改后，其他线程可以立刻看到
• 有序性：程序的执行顺序会按照代码的先后顺序执行。

实现原理

• Java虚拟机则是通过进入和退出Monitor对象来实现方法同步和代码块同步的
• 同步代码块的实现是由monitorenter 和monitorexit指令完成的，其中monitorenter指令所在的位置是同步代码块开始的位置，第一个monitorexit指令是用于正常结束同步代码块的指令

Java 并发编程基础

• 线程作为操作系统调度的最小单元，多个线程能够同时执行，这将显著提升程序性能，在多核环境中表现得更加明显。
• 过多地创建线程和对线程的不当管理也容易造成问题。

线程简介

• 操作系统调度的最小单元是线程，也叫轻量级进程（Light Weight Process），在一个进程里可以创建多个线程，这些线程都拥有各自的计数器、堆栈和局部变量等属性，并且能够访问共享的内存变量。处理器在这些线程上高速切换，让使用者感觉到这些线程在同时执行。

使用多线程的原因

• 提升处理器的使用效率

• 异步处理请求，提升系统响应速度提升用户体验

• Java 为多线程编程提供了良好、考究并且一致的编程模型，使开发人员能够更加专

  注于问题的解决

线程优先级

• 线程优先级的范围从1-10，默认优先级是5

• 针对频繁阻塞（休眠或者 I/O 操作）的线程需要设置较高优先级，而偏重计算（需要较多 CPU

  时间或者偏运算）的线程则设置较低的优先级，确保处理器不会被独占。

• 线程优先级不能作为程序正确性的依赖

线程的状态

• 测试.java

public class ThreadTest {

    public static void main(String[] args) {
        new Thread(new TimeWaiting(), "超时等待线程").start();
        new Thread(new Waiting(), "等待线程").start();
        // 使用两个 Blocked 线程，一个获取锁成功，另一个被阻塞
        new Thread(new Blocked(), "阻塞线程1").start();
        new Thread(new Blocked(), "阻塞线程2").start();
    }

    // 该线程不断地进行睡眠
    static class TimeWaiting implements Runnable {
        private static final Logger logger = LoggerFactory.getLogger(TimeWaiting.class);
        @Override
        public void run() {
            while (true) {
                logger.info(Thread.currentThread().toString());
                second(100);
            }
        }
    }

    // 该线程在 Waiting.class 实例上等待
    static class Waiting implements Runnable {
        private static final Logger logger = LoggerFactory.getLogger(Waiting.class);

        @Override
        public void run() {
            while (true) {
                synchronized (Waiting.class) {
                    try {
                        logger.info(Thread.currentThread().toString());
                        Waiting.class.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    // 该线程在 Blocked.class 实例上加锁后，不会释放该锁
    static class Blocked implements Runnable {
        private static final Logger logger = LoggerFactory.getLogger(Blocked.class);

        public void run() {
            synchronized (Blocked.class) {
                while (true) {
                    logger.info(Thread.currentThread().toString());
                    second(100);
                }
            }
        }
    }

    public static final void second(long seconds) {
        try {
            TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException e) {
        }
    }
}

• 运行结果

[超时等待线程] INFO util.thread.ThreadTest$TimeWaiting - Thread[超时等待线程,5,main]
[等待线程] INFO util.thread.ThreadTest$Waiting - Thread[等待线程,5,main]
[阻塞线程1] INFO util.thread.ThreadTest$Blocked - Thread[阻塞线程1,5,main]

• Jstack查看线程信息

// 线程被阻塞
"阻塞???程2" #14 prio=5 os_prio=0 tid=0x00000000282d2800 nid=0x231a0 waiting for monitor entry [0x0000000028f0f000]
   java.lang.Thread.State: BLOCKED (on object monitor)
        at util.thread.ThreadTest$Blocked.run(ThreadTest.java:61)
        - waiting to lock <0x00000007167efbb8> (a java.lang.Class for util.thread.ThreadTest$Blocked)
        at java.lang.Thread.run(Thread.java:748)
// 线程获取到了 Blocked.class 的锁
"阻塞线程1" #13 prio=5 os_prio=0 tid=0x00000000282d2000 nid=0x25670 waiting on condition [0x0000000028e0f000]
   java.lang.Thread.State: TIMED_WAITING (sleeping)
        at java.lang.Thread.sleep(Native Method)
        at java.lang.Thread.sleep(Thread.java:340)
        at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
        at util.thread.ThreadTest.second(ThreadTest.java:70)
        at util.thread.ThreadTest$Blocked.run(ThreadTest.java:62)
        - locked <0x00000007167efbb8> (a java.lang.Class for util.thread.ThreadTest$Blocked)
        at java.lang.Thread.run(Thread.java:748)
 // 等待线程 线程在 Waiting 实例上等待
"等待线程" #12 prio=5 os_prio=0 tid=0x00000000282d1000 nid=0x24514 in Object.wait() [0x0000000028d0e000]
   java.lang.Thread.State: WAITING (on object monitor)
        at java.lang.Object.wait(Native Method)
        - waiting on <0x00000007167ecd70> (a java.lang.Class for util.thread.ThreadTest$Waiting)
        at java.lang.Object.wait(Object.java:502)
        at util.thread.ThreadTest$Waiting.run(ThreadTest.java:45)
        - locked <0x00000007167ecd70> (a java.lang.Class for util.thread.ThreadTest$Waiting)
        at java.lang.Thread.run(Thread.java:748)
// 超时等待
"超时等待线程" #11 prio=5 os_prio=0 tid=0x00000000282cf800 nid=0x254d8 waiting on condition [0x0000000028c0e000]
   java.lang.Thread.State: TIMED_WAITING (sleeping)
        at java.lang.Thread.sleep(Native Method)
        at java.lang.Thread.sleep(Thread.java:340)
        at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
        at util.thread.ThreadTest.second(ThreadTest.java:70)
        at util.thread.ThreadTest$TimeWaiting.run(ThreadTest.java:30)
        at java.lang.Thread.run(Thread.java:748)

线程状态变迁

• 线程创建之后，调用 start()方法开始运行。当线程执行 wait()方法之后，线程进入等待状态。进入等待状态的线程需要依靠其他线程的通知才能够返回到运行状态，而超时等待状态相当于在等待状态的基础上增加了超时限制，也就是超时时间到达时将会返回到运行状态。当线程调用同步方法时，在没有获取到锁的情况下，线程将会进入到阻塞状态。线程在执行 Runnable 的 run()方法之后将会进入到终止状态。
  
• Java 将操作系统中的运行和就绪两个状态合并称为运行状态。
• 阻塞状态是线程阻塞在进入 synchronized 关键字修饰的方法或代码块（获取锁）时的状态，但是阻塞在 java.concurrent 包中 Lock 接口的线程状态却是等待状态，因为 java.concurrent 包中Lock 接口对于阻塞的实现均使用了 LockSupport 类中的相关方法。

Daemon 线程

• Daemon 属性需要在启动线程之前设置，不能在启动线程之后设置。
• Daemon 线程被用作完成支持性工作，但是在 Java 虚拟机退出时 Daemon 线程中的 finally块并不一定会执行。

启动和终止线程

构造线程

• 构造一个线程对象
• 提供线程所属的线程组、线程优先级、是否是 Daemon 等属性

private void init(ThreadGroup g, Runnable target, String name,
                      long stackSize, AccessControlContext acc,
                      boolean inheritThreadLocals) {
}

• 一个新构造的线程对象是由其 parent 线程来进行空间分配的，而child 线程继承了 parent 是否为 Daemon、优先级和加载资源的 contextClassLoader 以及可继承的 ThreadLocal，同时还会分配一个唯一的 ID 来标识这个 child 线程。至此，一个能够运行的线程对象就初始化好了，在堆内存中等待着运行。

启动线程

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

new Thread(new TimeWaiting(), "超时等待线程").start();

理解中断

• 线程的一个标识位属性，它表示一个运行中的线程是否被其他线程进行了中断操作。中断好比其他线程对该线程打了个招呼，其他线程通过调用该线程的interrupt()方法对其进行中断操作。

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

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

安全地终止线程

• 通过中断操作和 cancel()方法均可使线程得以终止。这种通过标识位或者中断操作的方式能够使线程在终止时有机会去清理资源，而不是武断地将线程停止，因此这种终止线程的做法显得更加安全和优雅。

@Test
public void test1() throws InterruptedException {
    Thread thread = new Thread(() -> {
        while (!Thread.currentThread().isInterrupted()) {
            logger.info("Current date:{}", System.currentTimeMillis());
        }
    });
    thread.start();
    Thread.sleep(3000);
    thread.interrupt();
    if(thread.isInterrupted()){
        logger.info("Thread was interrupted..");
    }
    Thread.sleep(3000);
}

线程间通信

volatile 和 synchronized 关键字

• 关键字volatile可以用来修饰字段（成员变量），就是告知程序任何对该变量的访问均需要从共享内存中获取，而对它的改变必须同步刷新回共享内存，它能保证所有线程对变量访问的可见性。
• 关键字synchronized可以修饰方法或者以同步块的形式来进行使用，它主要确保多个线程在同一个时刻，只能有一个线程处于方法或者同步块中，它保证了线程对变量访问的可见性和排他性。

等待/通知机制

• 一个线程修改一个对象的值，另一个线程感知变化。线程A调用对象O的wait()进入等待状态，另一个线程B调用对象O的notify()或者 notifuAll()方法，线程A 收到通知后从对象O 的wait()方法返回，执行后续操作。两个线程通过对象O来完成交互，而对象上的wait()和notify/notifyAll()的关系就如同开关信号一样，用来完成等待方和通知方之间的交互工作。
• 等待通知相关方法如下
  
• Consume.java

public class Consume {

    private static final Logger logger = LoggerFactory.getLogger(Consume.class);

    private final Object lockValue;

    public Consume(Object object) {
        this.lockValue = object;
    }

    /**
     * 生产者赋值
     */
    public void getValue() {
        synchronized (lockValue) {
            if (ObjectUtils.isEmpty(ProductConsumeValue.value)) {
                try {
                    lockValue.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            logger.info("Consume :{}", ProductConsumeValue.value);
            ProductConsumeValue.value = "";
            lockValue.notifyAll();
        }
    }
}

• Product.java

public class Product {

    private static final Logger logger = LoggerFactory.getLogger(Consume.class);

    private Object lockValue;

    public Product(Object lockValue) {
        this.lockValue = lockValue;
    }

    /**
     * 生产者赋值
     */
    public void setValue() {
        synchronized (lockValue) {
            if (!ObjectUtils.isEmpty(ProductConsumeValue.value)) {
                try {
                    lockValue.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            ProductConsumeValue.value = System.currentTimeMillis() + "_" + System.nanoTime();
            logger.info("Product :{}", ProductConsumeValue.value);
            lockValue.notify();
        }
    }
}

• Test.java

public static void main(String[] args) {
    String value = "";
    Product product = new Product(value);
    Consume consume = new Consume(value);
    ProductThread productThread = new ProductThread(product);
    ConsumerThread consumerThread = new ConsumerThread(consume);
    productThread.start();
    consumerThread.start();
}

等待方遵循如下原则

• 获取对象的锁。

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

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

  对应的伪代码如下:

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

通知方遵循如下原则

• 获得对象的锁。
• 改变条件
• 通知所有等待在对象上的线程。

​ 对应的伪代码如下：

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

管道输入/输出流

• 管道输入/输出流和普通的文件输入/输出流或者网络输入/输出流不同之处在于，它 主要用于线程之间的数据传输，而传输的媒介为内存。管道输入/输出流主要包括了如下4 种具体实现：PipedOutputStream、PipedInputStream、PipedReader 和 PipedWriter，前两种面向字节，而后两种面向字符。

Thread.join()的使用

• 如果主线程处理完其他的事务后，需要用到子线程的处理结果，也就是主线程需要等待子线程执行完成之后再结束，这个时候就要用到join()方法了。

• 如果一个线程 A 执行了 thread.join()语句，其含义是：当前线程 A 等待 thread 线程终止之后才从 thread.join()返回。

public class ThreadJoinTest {

    private static final Logger logger = LoggerFactory.getLogger(ThreadJoinTest.class);

    /**
     * 如果主线程处理完其他的事务后，需要用到子线程的处理结果，也就是主线程需要等待子线程执行完成之后再结束，这个时候就要用到 join() 方法了。
     *
     * @param args args
     * @throws InterruptedException 中断异常
     */
    public static void main(String[] args) throws InterruptedException {
        Thread currentThread = Thread.currentThread();
        for (int i = 0; i < 10; i++) {
            JoinThreadTest joinTestTread = new JoinThreadTest(currentThread);
            Thread thread = new Thread(joinTestTread, "线程 " + i);
            thread.start();
            currentThread = thread;
        }
        Thread.sleep(5000);
    }

    private static class JoinThreadTest implements Runnable {
        private final Thread thread;

        private JoinThreadTest(Thread currentThread) {
            thread = currentThread;
        }

        @Override
        public void run() {
            try {
                thread.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            logger.info("当前线程：{}", Thread.currentThread().getName());
        }
    }
}

ThreadLocal 的使用

• 线程变量，每个线程可以根据一个 ThreadLocal 对象查询到绑定在这个线程上的一个值。可以通过 set(T)方法来设置一个值，在当前线程下再通过 get()方法获取到原先设置的值。

public class ThreadLocalTest {

    private static final Logger logger = Logger.getLogger(String.valueOf(ThreadLocalTest.class));

    /**
     * 当使用 ThreadLocal 维护变量时，ThreadLocal 为每个使用该变量的线程提供独立的变量副本，
     * 所以每一个线程都可以独立地改变自己的副本，而不会影响其它线程所对应的副本。
     * 为了避免重复创建TSO（thread specific object，即与线程相关的变量） 使用 static final 修饰
     */
    private static final ThreadLocal<Map<String, String>> THREAD_LOCAL_MAP = new ThreadLocal<>();

    @Test
    public void test1() {
        Map<String, String> map = new HashMap<>();
        map.put("methodTest", "张三");
        map.put("test2", "李四");
        THREAD_LOCAL_MAP.set(map);
        getThreadLocalMap();
        THREAD_LOCAL_MAP.remove();
    }

    private void getThreadLocalMap() {
        Map<String, String> map = THREAD_LOCAL_MAP.get();
        logger.info(String.valueOf(map));
    }
}