啃碎并发（六）：Java线程同步与实现

时间 2019-12-07

标签并发 java 线程同步实现栏目 Java 繁體版

原文原文链接

0 前言

为什么要使用Java线程同步？ Java容许多线程并发控制，当多个线程同时操做一个可共享的资源变量时，将会致使数据不许确，相互之间产生冲突，所以加入同步锁以免在该线程没有完成操做以前，被其余线程的调用，从而保证了该变量的惟一性和准确性。java

但其并发编程的根本，就是使线程间进行正确的通讯。其中两个比较重要的关键点，以下：编程

线程通讯：重点关注线程同步的几种方式；

正确通讯：重点关注是否有线程安全问题；

Java中提供了不少线程同步操做，好比：synchronized关键字、wait/notifyAll、ReentrantLock、Condition、一些并发包下的工具类、Semaphore，ThreadLocal、AbstractQueuedSynchronizer等。本文主要说明一下这几种同步方式的使用及优劣。安全

1 ReentrantLock可重入锁

自JDK5开始，新增了Lock接口以及它的一个实现类ReentrantLock。ReentrantLock可重入锁是J.U.C包内置的一个锁对象，能够用来实现同步，基本使用方法以下：bash

public class ReentrantLockTest {

    private ReentrantLock lock = new ReentrantLock();

    public void execute() {
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + " do something synchronize");
            try {
                Thread.sleep(5000l);
            } catch (InterruptedException e) {
                System.err.println(Thread.currentThread().getName() + " interrupted");
                Thread.currentThread().interrupt();
            }
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                reentrantLockTest.execute();
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                reentrantLockTest.execute();
            }
        });
        thread1.start();
        thread2.start();
    }
}
复制代码

上面例子表示 同一时间段只能有1个线程执行execute方法，输出以下：多线程

Thread-0 do something synchronize
// 隔了5秒钟 输入下面
Thread-1 do something synchronize
复制代码

可重入锁中可重入表示的意义在于 对于同一个线程，能够继续调用加锁的方法，而不会被挂起。可重入锁内部维护一个计数器，对于同一个线程调用lock方法，计数器+1，调用unlock方法，计数器-1。并发

举个例子再次说明一下可重入的意思：在一个加锁方法execute中调用另一个加锁方法anotherLock并不会被挂起，能够直接调用(调用execute方法时计数器+1，而后内部又调用了anotherLock方法，计数器+1，变成了2)：dom

public void execute() {
    lock.lock();
    try {
        System.out.println(Thread.currentThread().getName() + " do something synchronize");
        try {
            anotherLock();
            Thread.sleep(5000l);
        } catch (InterruptedException e) {
            System.err.println(Thread.currentThread().getName() + " interrupted");
            Thread.currentThread().interrupt();
        }
    } finally {
        lock.unlock();
    }
}

public void anotherLock() {
    lock.lock();
    try {
        System.out.println(Thread.currentThread().getName() + " invoke anotherLock");
    } finally {
        lock.unlock();
    }
}
复制代码

输出：ide

Thread-0 do something synchronize
Thread-0 invoke anotherLock
// 隔了5秒钟 输入下面
Thread-1 do something synchronize
Thread-1 invoke anotherLock
复制代码

2 synchronized

synchronized跟ReentrantLock同样，也支持可重入锁。可是它是 一个关键字，是一种语法级别的同步方式，称为内置锁：工具

public class SynchronizedKeyWordTest {

    public synchronized void execute() {
            System.out.println(Thread.currentThread().getName() + " do something synchronize");
        try {
            anotherLock();
            Thread.sleep(5000l);
        } catch (InterruptedException e) {
            System.err.println(Thread.currentThread().getName() + " interrupted");
            Thread.currentThread().interrupt();
        }
    }

    public synchronized void anotherLock() {
        System.out.println(Thread.currentThread().getName() + " invoke anotherLock");
    }

    public static void main(String[] args) {
        SynchronizedKeyWordTest reentrantLockTest = new SynchronizedKeyWordTest();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                reentrantLockTest.execute();
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                reentrantLockTest.execute();
            }
        });
        thread1.start();
        thread2.start();
    }

}
复制代码

输出结果跟ReentrantLock同样，这个例子说明内置锁能够做用在方法上。synchronized关键字也能够修饰静态方法，此时若是调用该静态方法，将会锁住整个类。性能

同步是一种高开销的操做，所以应该尽可能减小同步的内容。一般没有必要同步整个方法，使用synchronized代码块同步关键代码便可。

synchronized跟ReentrantLock相比，有几点局限性：

加锁的时候不能设置超时。ReentrantLock有提供tryLock方法，能够设置超时时间，若是超过了这个时间而且没有获取到锁，就会放弃，而synchronized却没有这种功能；

ReentrantLock可使用多个Condition，而synchronized却只能有1个

不能中断一个试图得到锁的线程；

ReentrantLock能够选择公平锁和非公平锁；

ReentrantLock能够得到正在等待线程的个数，计数器等；

因此，Lock的操做与synchronized相比，灵活性更高，并且Lock提供多种方式获取锁，有Lock、ReadWriteLock接口，以及实现这两个接口的ReentrantLock类、ReentrantReadWriteLock类。

关于Lock对象和synchronized关键字选择的考量：

最好两个都不用，使用一种java.util.concurrent包提供的机制，可以帮助用户处理全部与锁相关的代码。

若是synchronized关键字能知足用户的需求，就用synchronized，由于它能简化代码。

若是须要更高级的功能，就用ReentrantLock类，此时要注意及时释放锁，不然会出现死锁，一般在finally代码释放锁。

在性能考量上来讲，若是竞争资源不激烈，二者的性能是差很少的，而当竞争资源很是激烈时（即有大量线程同时竞争），此时Lock的性能要远远优于synchronized。因此说，在具体使用时要根据适当状况选择。

3 Condition条件对象

Condition条件对象的意义在于 对于一个已经获取Lock锁的线程，若是还须要等待其余条件才能继续执行的状况下，才会使用Condition条件对象。

Condition能够替代传统的线程间通讯，用await()替换wait()，用signal()替换notify()，用signalAll()替换notifyAll()。

为何方法名不直接叫wait()/notify()/nofityAll()？由于Object的这几个方法是final的，不可重写！

public class ConditionTest {

    public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        Condition condition = lock.newCondition();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                lock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + " run");
                    System.out.println(Thread.currentThread().getName() + " wait for condition");
                    try {
                        condition.await();
                        System.out.println(Thread.currentThread().getName() + " continue");
                    } catch (InterruptedException e) {
                        System.err.println(Thread.currentThread().getName() + " interrupted");
                        Thread.currentThread().interrupt();
                    }
                } finally {
                    lock.unlock();
                }
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                lock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + " run");
                    System.out.println(Thread.currentThread().getName() + " sleep 5 secs");
                    try {
                        Thread.sleep(5000l);
                    } catch (InterruptedException e) {
                        System.err.println(Thread.currentThread().getName() + " interrupted");
                        Thread.currentThread().interrupt();
                    }
                    condition.signalAll();
                } finally {
                    lock.unlock();
                }
            }
        });
        thread1.start();
        thread2.start();
    }
}
复制代码

这个例子中thread1执行到condition.await()时，当前线程会被挂起，直到thread2调用了condition.signalAll()方法以后，thread1才会从新被激活执行。

这里须要注意的是thread1调用Condition的await方法以后，thread1线程释放锁，而后立刻加入到Condition的等待队列，因为thread1释放了锁，thread2得到锁并执行，thread2执行signalAll方法以后，Condition中的等待队列thread1被取出并加入到AQS中，接下来thread2执行完毕以后释放锁，因为thread1已经在AQS的等待队列中，因此thread1被唤醒，继续执行。

传统线程的通讯方式，Condition均可以实现。Condition的强大之处在于它能够为多个线程间创建不一样的Condition。

注意，Condition是被绑定到Lock上的，要建立一个Lock的Condition必须用newCondition()方法。

4 wait&notify/notifyAll方式

Java线程的状态转换图与相关方法，以下：

在图中，红框标识的部分方法，能够认为已过期，再也不使用。上图中的方法可以参与到线程同步中的方法，以下：

wait、notify、notifyAll方法：线程中通讯可使用的方法。线程中调用了wait方法，则进入阻塞状态，只有等另外一个线程调用与wait同一个对象的notify方法。这里有个特殊的地方，调用wait或者notify，前提是须要获取锁，也就是说，须要在同步块中作以上操做。

wait/notifyAll方式跟ReentrantLock/Condition方式的原理是同样的。

Java中每一个对象都拥有一个内置锁，在内置锁中调用wait，notify方法至关于调用锁的Condition条件对象的await和signalAll方法。

public class WaitNotifyAllTest {

    public synchronized void doWait() {
        System.out.println(Thread.currentThread().getName() + " run");
        System.out.println(Thread.currentThread().getName() + " wait for condition");
        try {
            this.wait();
            System.out.println(Thread.currentThread().getName() + " continue");
        } catch (InterruptedException e) {
            System.err.println(Thread.currentThread().getName() + " interrupted");
            Thread.currentThread().interrupt();
        }
    }

    public synchronized void doNotify() {
        try {
            System.out.println(Thread.currentThread().getName() + " run");
            System.out.println(Thread.currentThread().getName() + " sleep 5 secs");
            Thread.sleep(5000l);
            this.notifyAll();
        } catch (InterruptedException e) {
            System.err.println(Thread.currentThread().getName() + " interrupted");
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        WaitNotifyAllTest waitNotifyAllTest = new WaitNotifyAllTest();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                waitNotifyAllTest.doWait();
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                waitNotifyAllTest.doNotify();
            }
        });
        thread1.start();
        thread2.start();
    }
}
复制代码

这里须要注意的是 调用wait/notifyAll方法的时候必定要得到当前线程的锁，不然会发生IllegalMonitorStateException异常。

join方法：该方法主要做用是在该线程中的run方法结束后，才往下执行。

package com.thread.simple;
 
public class ThreadJoin {
    public static void main(String[] args) {
        Thread thread= new Thread(new Runnable() {
              @Override
              public void run() {
                   System.err.println("线程"+Thread.currentThread().getId()+" 打印信息");
              }
        });
        thread.start();
 	
        try {
            thread.join();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.err.println("主线程打印信息");	
    }
}
复制代码

yield方法：线程自己的调度方法，使用时线程能够在run方法执行完毕时，调用该方法，告知线程已能够出让CPU资源。

public class Test1 {  
    public static void main(String[] args) throws InterruptedException {  
        new MyThread("低级", 1).start();  
        new MyThread("中级", 5).start();  
        new MyThread("高级", 10).start();  
    }  
}  

class MyThread extends Thread {  
    public MyThread(String name, int pro) {  
        super(name);// 设置线程的名称 
        this.setPriority(pro);// 设置优先级 
    }  

    @Override  
    public void run() {  
        for (int i = 0; i < 30; i++) {  
            System.out.println(this.getName() + "线程第" + i + "次执行！");  
            if (i % 5 == 0)  
                Thread.yield();  
        }  
    }  
}  
复制代码

sleep方法：经过sleep(millis)使线程进入休眠一段时间，该方法在指定的时间内没法被唤醒，同时也不会释放对象锁；
```
/** * 能够明显看到打印的数字在时间上有些许的间隔 */
public class Test1 {  
    public static void main(String[] args) throws InterruptedException {  
        for(int i=0;i<100;i++){  
            System.out.println("main"+i);  
            Thread.sleep(100);  
        }  
    }  
} 
复制代码
```
sleep方法告诉操做系统 至少在指定时间内不需为线程调度器为该线程分配执行时间片，并不释放锁（若是当前已经持有锁）。实际上，调用sleep方法时并不要求持有任何锁。

因此，sleep方法并不须要持有任何形式的锁，也就不须要包裹在synchronized中。

5 ThreadLocal

ThreadLocal是一种把变量放到线程本地的方式来实现线程同步的。好比：SimpleDateFormat不是一个线程安全的类，可使用ThreadLocal实现同步，以下：

public class ThreadLocalTest {

    private static ThreadLocal<SimpleDateFormat> dateFormatThreadLocal = new ThreadLocal<SimpleDateFormat>() {
        @Override
        protected SimpleDateFormat initialValue() {
            return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        }
    };

    public static void main(String[] args) {
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                Date date = new Date();
                System.out.println(dateFormatThreadLocal.get().format(date));
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                Date date = new Date();
                System.out.println(dateFormatThreadLocal.get().format(date));
            }
        });
        thread1.start();
        thread2.start();
    }
}
复制代码

为什么SimpleDateFormat不是线程安全的类？具体请参考：

https://blog.csdn.net/zdp072/article/details/41044059

https://blog.csdn.net/zq602316498/article/details/40263083

ThreadLocal与同步机制的对比选择：

ThreadLocal与同步机制都是 为了解决多线程中相同变量的访问冲突问题。

前者采用以 "空间换时间" 的方法，后者采用以 "时间换空间" 的方式。

6 volatile修饰变量

volatile关键字为域变量的访问提供了一种免锁机制，使用volatile修饰域至关于告诉虚拟机该域可能会被其余线程更新，所以每次使用该域就要从新计算，而不是使用寄存器中的值，volatile不会提供任何原子操做，它也不能用来修饰final类型的变量。

//只给出要修改的代码，其他代码与上同
public class Bank {
    //须要同步的变量加上volatile
    private volatile int account = 100;
    public int getAccount() {
        return account;
    }
    //这里再也不须要synchronized 
    public void save(int money) {
        account += money;
    }
｝
复制代码

多线程中的非同步问题主要出如今对域的读写上，若是让域自身避免这个问题，则就不须要修改操做该域的方法。用final域，有锁保护的域和volatile域能够避免非同步的问题。

7 Semaphore信号量

Semaphore信号量被用于控制特定资源在同一个时间被访问的个数。相似链接池的概念，保证资源能够被合理的使用。可使用构造器初始化资源个数：

public class SemaphoreTest {

    private static Semaphore semaphore = new Semaphore(2);

    public static void main(String[] args) {
        for(int i = 0; i < 5; i ++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + " " + new Date());
                        Thread.sleep(5000l);
                        semaphore.release();
                    } catch (InterruptedException e) {
                        System.err.println(Thread.currentThread().getName() + " interrupted");
                    }
                }
            }).start();
        }
    }
}
复制代码

输出：

Thread-1 Mon Apr 18 18:03:46 CST 2016
Thread-0 Mon Apr 18 18:03:46 CST 2016
Thread-3 Mon Apr 18 18:03:51 CST 2016
Thread-2 Mon Apr 18 18:03:51 CST 2016
Thread-4 Mon Apr 18 18:03:56 CST 2016
复制代码

8 并发包下的工具类

8.1 CountDownLatch

CountDownLatch是一个计数器，它的构造方法中须要设置一个数值，用来设定计数的次数。每次调用countDown()方法以后，这个计数器都会减去1，CountDownLatch会一直阻塞着调用await()方法的线程，直到计数器的值变为0。

public class CountDownLatchTest {

    public static void main(String[] args) {
        CountDownLatch countDownLatch = new CountDownLatch(5);
        for(int i = 0; i < 5; i ++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + " " + new Date() + " run");
                    try {
                        Thread.sleep(5000l);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    countDownLatch.countDown();
                }
            }).start();
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("all thread over");
    }
}
复制代码

输出：

Thread-2 Mon Apr 18 18:18:30 CST 2016 run
Thread-3 Mon Apr 18 18:18:30 CST 2016 run
Thread-4 Mon Apr 18 18:18:30 CST 2016 run
Thread-0 Mon Apr 18 18:18:30 CST 2016 run
Thread-1 Mon Apr 18 18:18:30 CST 2016 run
all thread over
复制代码

8.2 CyclicBarrier

CyclicBarrier阻塞调用的线程，直到条件知足时，阻塞的线程同时被打开。

调用await()方法的时候，这个线程就会被阻塞，当调用await()的线程数量到达屏障数的时候，主线程就会取消全部被阻塞线程的状态。

在CyclicBarrier的构造方法中，还能够设置一个barrierAction。在全部的屏障都到达以后，会启动一个线程来运行这里面的代码。

public class CyclicBarrierTest {

    public static void main(String[] args) {
        Random random = new Random();
        CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
        for(int i = 0; i < 5; i ++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    int secs = random.nextInt(5);
                    System.out.println(Thread.currentThread().getName() + " " + new Date() + " run, sleep " + secs + " secs");
                    try {
                        Thread.sleep(secs * 1000);
                        cyclicBarrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + " " + new Date() + " runs over");
                }
            }).start();
        }
    }
}
复制代码

相比CountDownLatch，CyclicBarrier是能够被循环使用的，并且遇到线程中断等状况时，还能够利用reset()方法，重置计数器，从这些方面来讲，CyclicBarrier会比CountDownLatch更加灵活一些。

9 使用原子变量实现线程同步

有时须要使用线程同步的根本缘由在于 对普通变量的操做不是原子的。那么什么是原子操做呢？

原子操做就是指将读取变量值、修改变量值、保存变量值当作一个总体来操做即-这几种行为要么同时完成，要么都不完成。

在java.util.concurrent.atomic包中提供了建立原子类型变量的工具类，使用该类能够简化线程同步。好比：其中AtomicInteger以原子方式更新int的值：

class Bank {
    private AtomicInteger account = new AtomicInteger(100);

    public AtomicInteger getAccount() {
        return account;
    }

    public void save(int money) {
        account.addAndGet(money);
    }
}
复制代码

10 AbstractQueuedSynchronizer

AQS是不少同步工具类的基础，好比：ReentrantLock里的公平锁和非公平锁，Semaphore里的公平锁和非公平锁，CountDownLatch里的锁等他们的底层都是使用AbstractQueuedSynchronizer完成的。

基于AbstractQueuedSynchronizer自定义实现一个独占锁：

public class MySynchronizer extends AbstractQueuedSynchronizer {

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

    @Override
    protected boolean tryRelease(int arg) {
        setState(0);
        setExclusiveOwnerThread(null);
        return true;
    }

    public void lock() {
        acquire(1);
    }

    public void unlock() {
        release(1);
    }

    public static void main(String[] args) {
        MySynchronizer mySynchronizer = new MySynchronizer();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                mySynchronizer.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + " run");
                    System.out.println(Thread.currentThread().getName() + " will sleep 5 secs");
                    try {
                        Thread.sleep(5000l);
                        System.out.println(Thread.currentThread().getName() + " continue");
                    } catch (InterruptedException e) {
                        System.err.println(Thread.currentThread().getName() + " interrupted");
                        Thread.currentThread().interrupt();
                    }
                } finally {
                    mySynchronizer.unlock();
                }
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                mySynchronizer.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + " run");
                } finally {
                    mySynchronizer.unlock();
                }
            }
        });
        thread1.start();
        thread2.start();
    }
}
复制代码

11 使用阻塞队列实现线程同步

前面几种同步方式都是基于底层实现的线程同步，可是在实际开发当中，应当尽可能远离底层结构。本节主要是使用LinkedBlockingQueue来实现线程的同步。

LinkedBlockingQueue是一个基于链表的队列，先进先出的顺序（FIFO），范围任意的blocking queue。

package com.xhj.thread;

import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;

/** * 用阻塞队列实现线程同步 LinkedBlockingQueue的使用 */
public class BlockingSynchronizedThread {
    /** * 定义一个阻塞队列用来存储生产出来的商品 */
    private LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
    /** * 定义生产商品个数 */
    private static final int size = 10;
    /** * 定义启动线程的标志，为0时，启动生产商品的线程；为1时，启动消费商品的线程 */
    private int flag = 0;

    private class LinkBlockThread implements Runnable {
        @Override
        public void run() {
            int new_flag = flag++;
            System.out.println("启动线程 " + new_flag);
            if (new_flag == 0) {
                for (int i = 0; i < size; i++) {
                    int b = new Random().nextInt(255);
                    System.out.println("生产商品：" + b + "号");
                    try {
                        queue.put(b);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                    System.out.println("仓库中还有商品：" + queue.size() + "个");
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            } else {
                for (int i = 0; i < size / 2; i++) {
                    try {
                        int n = queue.take();
                        System.out.println("消费者买去了" + n + "号商品");
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                    System.out.println("仓库中还有商品：" + queue.size() + "个");
                    try {
                        Thread.sleep(100);
                    } catch (Exception e) {
                        // TODO: handle exception
                    }
                }
            }
        }
    }

    public static void main(String[] args) {
        BlockingSynchronizedThread bst = new BlockingSynchronizedThread();
        LinkBlockThread lbt = bst.new LinkBlockThread();
        Thread thread1 = new Thread(lbt);
        Thread thread2 = new Thread(lbt);
        thread1.start();
        thread2.start();
    }
}
复制代码