Java多线程基础

一、线程概述

1. 并发与并行

并行：指在同一时刻，有多条指令在多个处理器上同时执行；

并发：指在同一时刻只能有一条指令执行，但是被快速轮转，表现为多线程。

2. 多线程编程的优点

进程之间不能共享内存，但是线程之间共享内存很容易；
对操作系统而言，线程的创建代价小，而进程的创建和销毁代价很高；

二、线程创建和启动

1. 继承Thread类创建线程类

定义Thead类的子类，重写run()方法；
创建Thead子类的实例，即创建了线程对象；
调用线程对象的start()方法来启动线程；

 1 package thread;
 2 
 3 public class FirstThread extends Thread{
 4 
 5     private int i = 0;
 6     @Override
 7     public void run() {
 8         
 9         for(;i<100;i++){
10             System.out.println(getName() + " " + i);
11         }
12     }
13     
14     public static void main(String[] args) {
15         for(int i=0;i<100;i++){
16             System.out.println(Thread.currentThread().getName() +  " " + i);
17             
18             if(i==20){
19                 new FirstThread().start();
20                 new FirstThread().start();
21             }
22         }
23     }
24 }

View Code

Thread.currentThread()，返回当前正在执行的线程对象；
getName()，返回调用该方法的线程名字；
多个线程中间无法共享线程类的实例变量；

2. 实现Runable接口创建线程类

定义Runable接口的实现类，重写run方法；
创建Runable实现类的实例，并以此实例作为Thread的target来创建Thread对象；
调用线程对象的start方法启动线程；

 1 package thread;
 2 
 3 public class SecondThread implements Runnable{
 4 
 5     private int i = 0;
 6     
 7     @Override
 8     public void run() {
 9         for(;i<100;i++){
10             System.out.println(Thread.currentThread().getName() + " " + i);
11         }
12     }
13     
14     public static void main(String[] args) {
15         
16         for(int i = 0;i<100;i++){
17             System.out.println(Thread.currentThread().getName() + " " + i);
18         
19             if( i == 20){
20                 SecondThread st = new SecondThread();
21                 new Thread(st,"新线程1").start();
22                 new Thread(st,"新线程2").start();
23             }
24         }
25         
26     }
27 }

View Code

由于实现Runable接口创建的线程，多个线程共享一个target类时，可以共享实例变量。

3. 使用Callable和Future创建线程

Callable接口是Runable接口的增强版，提供了一个call()方法作为线程执行体，但是比run方法更强大！

call方法可以有返回值
call方法可以声明抛出异常

创建并启动有返回值的线程的步骤：

创建Callable接口的实现类，并实现call方法；
使用FutureTask类来包装Callable对象，该FutureTask对象封装了该Callable对象的call方法的返回值；
使用FutureTask对象作为Thread对象的target创建并启动新线程；
调用FutureTask对象的get方法来获取子线程执行结束后的返回值；

FutureTask类的公共方法：

boolean cancel(boolean mayInterruptIfRunning)
V get()
V get(long timeout, TimeUnit unit)
boolean isCancelled()
boolean isDone()

 1 package thread;
 2 
 3 import java.util.concurrent.Callable;
 4 import java.util.concurrent.FutureTask;
 5 
 6 public class ThirdThread {
 7 
 8     public static void main(String[] args) {
 9         
10         ThirdThread rt = new ThirdThread();
11         
12         FutureTask<Integer> task = new FutureTask<>(
13                 (Callable<Integer>) () -> {
14                     int i=0;
15                     for(;i<100;i++){
16                         System.out.println(Thread.currentThread().getName()
17                                 + " 的循环变量i的值： " + i);
18                     }
19                     return i;
20                 }
21         );
22         
23         for(int i=0;i<100;i++){
24             System.out.println(Thread.currentThread().getName()
25                     + " 的循环变量i的值： " + i );
26             if( i== 20){
27                 new Thread(task, "有返回值的线程").start();
28             }
29         }
30         
31         try{
32             System.out.println("子线程的返回值" + task.get());
33         }catch (Exception e) {
34             e.printStackTrace();
35         }
36      }
37 }

View Code

三、线程的生命周期

Java线程的五种状态：新建，就绪，运行，阻塞，死亡。

1. 新建和就绪状态

new创建一个线程之后，线程就处于新建状态。

调用线程的start()方法之后，线程处于就绪状态。

注意：只能对新建的线程调用start()方法，否则引入IllegalThreadStateException异常。

2. 运行和阻塞状态

线程从运行状态进入阻塞状态：

线程调用sleep方法主动放弃所占用的处理器资源
线程调用了一个阻塞IO，等待IO结果
线程等待一个排它锁
线程等待一个通知
调用线程的suspend方法，主动挂起线程。

线程从阻塞状态进入就绪状态：

sleep方法经过了指定的时间；
线程IO结果返回；
线程获取了锁；
线程收到了通知;
挂起的线程被调用了resume方法恢复；

3. 线程死亡

有三种方式结束线程：

run和call方法执行完成，线程正常结束；
线程抛出了一个未捕获的异常；
直接调用线程的stop方法结束线程（容易死锁）；

四、控制线程

1. join线程

一个线程通过调用另一个线程的join方法，等待另外一个线程完成后，继续执行。

join()方法的三种重载：

join()
join(long millis)
join(long millis, int nanos):等待最长milis毫秒加nanos毫微秒

 1 package thread;
 2 
 3 public class JoinThread extends Thread{
 4 
 5     // 提供带参构造函数，为线程起名
 6     public JoinThread(String name) {
 7         super(name);
 8     }
 9     
10     @Override
11     public void run() {
12         
13         for(int i=0;i<100;i++) {
14             System.out.println(getName() + " " + i);
15         }
16     }
17     
18     public static void main(String[] args) throws Exception {
19         new JoinThread("新线程").start();
20         
21         for(int i=0;i<100;i++) {
22             if(i==20) {
23                 JoinThread jt = new JoinThread("被Join的线程");
24                 jt.start();
25                 //main线程调用了join方法，必须等待jt执行结束，才能继续执行
26                 jt.join();
27             }
28             System.out.println(Thread.currentThread().getName()
29                     + " " + i );
30         }
31     }
32 }

View Code

2. 后台线程

后台线程的特征：如果所有的前台线程全部死亡，后台线程自动死亡。

调用thread对象的setDeamon(True)方法，将线程设置为后台线程。

 1 package thread;
 2 
 3 public class DeamonThread  extends Thread{
 4 
 5     @Override
 6     public void run() {
 7         
 8         for(int i=0;i<1000;i++) {
 9             System.out.println(getName() + " " + i);
10         }
11     }
12     
13     public static void main(String[] args) {
14         
15         DeamonThread t = new DeamonThread();
16         t.setDaemon(true);
17         t.start();
18         for(int i=0;i<10;i++) {
19             System.out.println(Thread.currentThread().getName()
20                     + " " + i );
21         }
22     }
23 }

View Code

3. 线程睡眠：sleep

让正在运行的线程暂停一段时间，进入阻塞状态，需要调用Thread类的sleep方法来实现！

两种重载形式：

sleep(long millis)
sleep(long millis, int nanos)

 1 package thread;
 2 
 3 import java.util.Date;
 4 
 5 public class SleepTest {
 6 
 7     public static void main(String[] args) throws InterruptedException {
 8         for(int i=0;i<10;i++) {
 9             System.out.println("当前时间 ：" + new Date());
10             Thread.sleep(1000);
11         }
12     }
13 }

View Code

4. 线程让步：yield

yield方法是和sleep方法很类似的一个Thread类中的静态方法，作用是让正在进行的线程暂停，直接转为就绪状态，等待下次调度。

需要注意的是，当某个线程调用了yield方法之后，只有优先级与当前线程相同或者更高并处于就绪状态的线程才能获得执行机会。

5. 改变线程优先级

　　每个线程都有自己的优先级，子线程会继承父线程的优先级。

　　为线程设置优先级的方法：

setPriority(int newPriority)
MIN_PRIORITY、NORM_PRIORITY、MAX_PRIORITY

五、线程同步

1. 锁机制

一般使用synchronized锁，或者j.u.c包中提供的锁机制来实现线程同步：

jdk5中提供的j.u.c包提供了重入锁机制：

java.util.concurrent.locks.ReentrantLock

构造方法：

ReentrantLock()
ReentrantLock(boolean fair): 公平锁 ——> 按照FIFO的机制，来实现公平竞争。

2. 条件对象

为什么需要条件对象？

有时候虽然线程进入临界区，却只能在满足某一条件之后才能执行，因此需要使用一个条件变量来管理已经获得一个锁但是却不能做有用工作的线程。

条件对象用来管理那些已经进入保护代码段，但是还不能运行的线程。

条件对象（与对象的wait、signal、signalAll功能类似）的使用方法：

await() 会放弃锁
signal()
signalAll()

await()的用法，一般为：

while(!(ok to proceed)){
    condition.await();          
}

在程序中应该使用Lock、Condition，还是synchronized？

最好都不使用。最好使用j.u.c包中的机制来实现线程同步。优先使用synchronized，如果Lock、Condition更适合，就使用Lock、Condition。

3. 读写锁

使用读写锁的步骤：

　　1. 构造一个ReentrantReadWriteLock对象

　　　 private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

　　2. 抽取读锁和写锁

　　 private Lock readLock = rwl.readLock();

private Lock writeLock = rwl.writeLock();

　　3. 对所有的访问者加读锁

　　readLock.lock();
        try {
            // do read
        } finally {
            readLock.unlock();
        }

　　4. 对所有的修改者加写锁

　　writeLock.lock();
        
        try {
            // do something
        } finally {
            writeLock.unlock();
        }

六. 阻塞队列

1. 阻塞队列基础

上述同步互斥方法属于Java底层机制，在实际开发中，应该尽量避免使用底层技术，而使用高效的其他结构来实现。

使用队列可以安全的从一个线程向另一个线程传递数据。

阻塞队列提供了四种处理方法:

方法处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(time,unit)
检查方法	element()	peek()	不可用	不可用

抛出异常：是指当阻塞队列满时候，再往队列里插入元素，会抛出IllegalStateException("Queue full")异常。当队列为空时，从队列里获取元素时会抛出NoSuchElementException异常。
返回特殊值：插入方法会返回是否成功，成功则返回true。移除方法，则是从队列里拿出一个元素，如果没有则返回null
一直阻塞：当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到拿到数据，或者响应中断退出。当队列空时，消费者线程试图从队列里take元素，队列也会阻塞消费者线程，直到队列可用。
超时退出：当阻塞队列满时，队列会阻塞生产者线程一段时间，如果超过一定的时间，生产者线程就会退出。

2. JDK中的阻塞队列

JDK7提供了7个阻塞队列。分别是

ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。
LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列。
PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。
DelayQueue：一个使用优先级队列实现的无界阻塞队列。
SynchronousQueue：一个不存储元素的阻塞队列。
LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。默认情况下不保证访问者公平的访问队列，所谓公平访问队列是指阻塞的所有生产者线程或消费者线程，当队列可用时，可以按照阻塞的先后顺序访问队列，即先阻塞的生产者线程，可以先往队列里插入元素，先阻塞的消费者线程，可以先从队列里获取元素。通常情况下为了保证公平性会降低吞吐量。我们可以使用以下代码创建一个公平的阻塞队列：

ArrayBlockingQueue fairQueue = new  ArrayBlockingQueue(1000,true);

访问者的公平性是使用可重入锁实现的，代码如下：

public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
}

LinkedBlockingQueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。

PriorityBlockingQueue是一个支持优先级的无界队列。默认情况下元素采取自然顺序排列，也可以通过比较器comparator来指定元素的排序规则。元素按照升序排列。

DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。我们可以将DelayQueue运用在以下应用场景：

缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。
定时任务调度。使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，从比如TimerQueue就是使用DelayQueue实现的。

队列中的Delayed必须实现compareTo来指定元素的顺序。比如让延时时间最长的放在队列的末尾。实现代码如下：

public int compareTo(Delayed other) {
           if (other == this) // compare zero ONLY if same object
                return 0;
            if (other instanceof ScheduledFutureTask) {
                ScheduledFutureTask x = (ScheduledFutureTask)other;
                long diff = time - x.time;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
       else if (sequenceNumber < x.sequenceNumber)
                    return -1;
                else
                    return 1;
            }
            long d = (getDelay(TimeUnit.NANOSECONDS) -
                      other.getDelay(TimeUnit.NANOSECONDS));
            return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
        }

如何实现Delayed接口

我们可以参考ScheduledThreadPoolExecutor里ScheduledFutureTask类。这个类实现了Delayed接口。首先：在对象创建的时候，使用time记录前对象什么时候可以使用，代码如下：

ScheduledFutureTask(Runnable r, V result, long ns, long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
}

然后使用getDelay可以查询当前元素还需要延时多久，代码如下：

public long getDelay(TimeUnit unit) {
            return unit.convert(time - now(), TimeUnit.NANOSECONDS);
        }

通过构造函数可以看出延迟时间参数ns的单位是纳秒，自己设计的时候最好使用纳秒，因为getDelay时可以指定任意单位，一旦以纳秒作为单位，而延时的时间又精确不到纳秒就麻烦了。使用时请注意当time小于当前时间时，getDelay会返回负数。

如何实现延时队列

延时队列的实现很简单，当消费者从队列里获取元素时，如果元素没有达到延时时间，就阻塞当前线程。

long delay = first.getDelay(TimeUnit.NANOSECONDS);
                    if (delay <= 0)
                        return q.poll();
                    else if (leader != null)
                        available.await();

SynchronousQueue是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能继续添加元素。SynchronousQueue可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合于传递性场景,比如在一个线程中使用的数据，传递给另外一个线程使用，SynchronousQueue的吞吐量高于LinkedBlockingQueue 和 ArrayBlockingQueue。

LinkedTransferQueue是一个由链表结构组成的无界阻塞TransferQueue队列。相对于其他阻塞队列，LinkedTransferQueue多了tryTransfer和transfer方法。

transfer方法。如果当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法时），transfer方法可以把生产者传入的元素立刻transfer（传输）给消费者。如果没有消费者在等待接收元素，transfer方法会将元素存放在队列的tail节点，并等到该元素被消费者消费了才返回。transfer方法的关键代码如下：

Node pred = tryAppend(s, haveData);
return awaitMatch(s, pred, e, (how == TIMED), nanos);

第一行代码是试图把存放当前元素的s节点作为tail节点。第二行代码是让CPU自旋等待消费者消费元素。因为自旋会消耗CPU，所以自旋一定的次数后使用Thread.yield()方法来暂停当前正在执行的线程，并执行其他线程。

tryTransfer方法。则是用来试探下生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，则返回false。和transfer方法的区别是tryTransfer方法无论消费者是否接收，方法立即返回。而transfer方法是必须等到消费者消费了才返回。

对于带有时间限制的tryTransfer(E e, long timeout, TimeUnit unit)方法，则是试图把生产者传入的元素直接传给消费者，但是如果没有消费者消费该元素则等待指定的时间再返回，如果超时还没消费元素，则返回false，如果在超时时间内消费了元素，则返回true。

LinkedBlockingDeque是一个由链表结构组成的双向阻塞队列。所谓双向队列指的你可以从队列的两端插入和移出元素。双端队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。相比其他的阻塞队列，LinkedBlockingDeque多了addFirst，addLast，offerFirst，offerLast，peekFirst，peekLast等方法，以First单词结尾的方法，表示插入，获取（peek）或移除双端队列的第一个元素。以Last单词结尾的方法，表示插入，获取或移除双端队列的最后一个元素。另外插入方法add等同于addLast，移除方法remove等效于removeFirst。但是take方法却等同于takeFirst，不知道是不是Jdk的bug，使用时还是用带有First和Last后缀的方法更清楚。

在初始化LinkedBlockingDeque时可以设置容量防止其过渡膨胀。另外双向阻塞队列可以运用在“工作窃取”模式中。

3. 阻塞队列的实现原理

如果队列是空的，消费者会一直等待，当生产者添加元素时候，消费者是如何知道当前队列有元素的呢？如果让你来设计阻塞队列你会如何设计，让生产者和消费者能够高效率的进行通讯呢？让我们先来看看JDK是如何实现的。

使用通知模式实现。所谓通知模式，就是当生产者往满的队列里添加元素时会阻塞住生产者，当消费者消费了一个队列中的元素后，会通知生产者当前队列可用。通过查看JDK源码发现ArrayBlockingQueue使用了Condition来实现，代码如下：

private final Condition notFull;
private final Condition notEmpty;

public ArrayBlockingQueue(int capacity, boolean fair) {
        //省略其他代码
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            insert(e);
        } finally {
            lock.unlock();
        }
}

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return extract();
  } finally {
            lock.unlock();
        }
}

private void insert(E x) {
        items[putIndex] = x;
        putIndex = inc(putIndex);
        ++count;
        notEmpty.signal();
    }

当我们往队列里插入一个元素时，如果队列不可用，阻塞生产者主要通过LockSupport.park(this);来实现

public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)

　　　　　　　　reportInterruptAfterWait(interruptMode);
        }

继续进入源码，发现调用setBlocker先保存下将要阻塞的线程，然后调用unsafe.park阻塞当前线程。

public static void park(Object blocker) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        unsafe.park(false, 0L);
        setBlocker(t, null);
    }

unsafe.park是个native方法，代码如下：

public native void park(boolean isAbsolute, long time);

park这个方法会阻塞当前线程，只有以下四种情况中的一种发生时，该方法才会返回。

与park对应的unpark执行或已经执行时。注意：已经执行是指unpark先执行，然后再执行的park。
线程被中断时。
如果参数中的time不是零，等待了指定的毫秒数时。
发生异常现象时。这些异常事先无法确定。

六、线程通信

七、线程相关类