1.并发编程基础

作者: torres1 | 来源:发表于2020-01-06 18:45 被阅读0次

1.线程相关定义

现代操作系统在运行一个程序时，会为其创建一个进程。例如，启动一个Java程序，操作系统就会创建一个Java进程。线程是现代操作系统调度的最小单元，也叫轻量级进程，在一个进程里可以创建多个线程，这些线程都拥有各自的计算器、堆栈和局部变量等属性，并且能够访问共享的内存变量。处理器在这些线程上高速切换，让使用者感觉到这些线程在同时执行。

2.线程的状态

Java线程在运行的生命周期中可能处于下表所示的6种不同的状态，在给定的一个时刻，线程只能处于其中的一个状态。

状态名称	状态说明
NEW	初始状态，线程被构建，但是还没有调用start()方法
RUNNABLE	运行状态，Java线程将操作系统中的就绪和运行两种状态笼统地称作“运行中”
WAITING	等待状态，表示线程进入等待状态，进入该状态表示当前线程需要等待其他线程做出一些特定动作（通知或中断）
BLOCKED	阻塞状态，表示线程阻塞于锁
TIME_WAITING	超时等待状态，该状态不同于WAITING，它是可以在指定的时间自行返回的
TERMINATED	终止状态，表示当前线程已经执行完毕

线程状态迁移图：

image.png

3.线程的启动与终止

3.1线程的两种创建方式

1.继承Thread

public class ThreadOne extends Thread {

    @Override
    public void run() {
        System.out.println("ThreadOne run.....");
    }
}

2.实现Runnable接口

public class RunnableTwo implements Runnable {
    @Override
    public void run() {
        System.out.println("RunnnableTwo run...");
    }
}
//运行
    public static void main(String[] args) {
        ThreadOne threadOne = new ThreadOne();
        threadOne.start();
        Thread threadTwo = new Thread(new RunnableTwo());
        threadTwo.start();
    }

3.2线程的中断

3.2.1什么是线程中断?

线程中断即线程运行过程中被其他线程给打断了，它与 stop 最大的区别是：stop 是由系统强制终止线程，而线程中断则是给目标线程发送一个中断信号，如果目标线程没有接收线程中断的信号并结束线程，线程则不会终止，具体是否退出或者执行其他逻辑由目标线程决定。

3.2.2三个重要的方法

1.java.lang.Thread#interrupt()
给目标线程发送一个中断信号，线程被打上中断标记。
2.java.lang.Thread#isInterrupted()
判断目标线程是否处于中断状态，不会清除中断标记。
3.java.lang.Thread#interrupted（静态方法）
判断当前线程是否中断，同时清除中断标记。

3.2.3如何安全的终止线程？

可以使用中断操作或者标志位的方式来控制是否需要停止任务并终止线程。
下面例子中，countThread，不断地进行变量累加，而主线程尝试进行中断操作。

public class CountRunner implements Runnable {
    private long i;
    private volatile boolean flag = true;

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

    public void cancel() {
        this.flag = false;
    }
}

public class Shutdown {

    public static void main(String[] args) throws Exception {
        CountRunner one = new CountRunner();
        Thread countThread = new Thread(one, "countThread");
        countThread.start();
        //休眠1s，感知中断而结束
        TimeUnit.SECONDS.sleep(1);
        countThread.interrupt();
        CountRunner two = new CountRunner();
        countThread = new Thread(two, "countThread");
        countThread.start();
        //休眠1s，感知标志位改变结束
        TimeUnit.SECONDS.sleep(1);
        two.cancel();
    }
}

输出

count =764735764
count =770902239

4.线程间的通信

4.1等待/通知相关方法

方法名称	描述
wait()	调用该方法的线程进入waiting状态，只有通过其他线程的通知或者中断才会返回，使用wait后，会释放对象的锁
wait(long)	等待指定时间，ms，没有通知就超时返回
wait(long,int)	对于超时时间更细粒度的控制，可以达到纳秒
notify()	通知一个在对象上等待的线程，使其从wait()方法中返回，而返回的前提是获取了对象的锁
notifyAll()	通知所有等待在该对象上的线程

等待/通知机制，是指一个线程A调用了对象O的wait()方法进入等待状态，而另一个线程B 调用了对象O的notify()或者notifyAll()方法，线程A收到通知后从对象O的wait()方法返回，进而执行后续操作。上述两个线程通过对象O来完成交互，而对象上的wait()和notify/notifyAll()的关系就如同开关信号一样，用来完成等待方和通知方之间的交互工作。
例子：

package wainotify;

import java.util.concurrent.TimeUnit;

public class WaitNotify {
    private static final Object lock = new Object();
    private static boolean flag = false;

    public static void main(String[] args) throws Exception {
        Thread waitThread = new Thread(new WaitThread());
        waitThread.start();
        TimeUnit.SECONDS.sleep(1);
        Thread notifyThread = new Thread(new NotifyThread());
        notifyThread.start();
    }

    static class WaitThread implements Runnable {

        @Override
        public void run() {
            synchronized (lock) {
                while (!flag) {
                    try {
                        System.out.println("WaitThread waiting ......");
                        lock.wait();
                    }catch (InterruptedException ex){
                    }
                }
                System.out.println("WaitThread start......");
            }
            System.out.println("WaitThread end");
        }
    }

    static class NotifyThread implements Runnable {

        @Override
        public void run() {
            synchronized (lock) {
                flag = true;
                System.out.println("NotifyThread notifyAll ......");
                lock.notifyAll();
            }
            System.out.println("NotifyThread end");
        }
    }
}

交互图

image.png

4.2等待/通知经典范式

等待方
1.获取对象的锁
2.如果条件不满足，则调用对象的wait方法，被通知后仍要检查
3.执行相应的逻辑

synchronized(对象){
while(!flag){
object.wait();
}
执行相应的逻辑
}

通知方
1.获取对象的锁
2.改变条件
3.通知所有等待在对象上的线程

synchronized(对象){
flag=true
object.notifyAll()
}

4.3 join

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

package join;

public class JoinMain {

    public static void main(String[] args) {
        Thread previous = Thread.currentThread();
        for (int i = 0; i < 10; i++) {
            Thread thread = new Thread(new Domino(previous), "thread_" + i);
            thread.start();
            previous = thread;
        }
        System.out.println(Thread.currentThread().getName() + "..." + "end");
    }


    static class Domino implements Runnable {
        private Thread previous;

        Domino(Thread previous) {
            this.previous = previous;
        }

        @Override
        public void run() {
            try {
                previous.join();
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread().getName() + "..." + "end");
        }
    }
}

join源码

    public final synchronized void join(long millis)
    throws InterruptedException {
        long base = System.currentTimeMillis();
        long now = 0;

        if (millis < 0) {
            throw new IllegalArgumentException("timeout value is negative");
        }

        if (millis == 0) {
            while (isAlive()) {
                wait(0);
            }
        } else {
            while (isAlive()) {
                long delay = millis - now;
                if (delay <= 0) {
                    break;
                }
                wait(delay);
                now = System.currentTimeMillis() - base;
            }
        }
    }

4.4ThreadLocal

当访问共享变量时，往往需要加锁来保证数据同步。一种避免使用同步的方式就是不共享数据。如果仅在单线程中访问数据，就不需要同步了。这种技术称为线程封闭。在Java语言中，提供了一些类库和机制来维护线程的封闭性，例如局部变量和ThreadLocal类。

package threadlocal;

public class ThreadLocalTest {

    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            Thread thread1 = new Thread(new Local1());
            Thread thread2 = new Thread(new local2());
            thread1.start();
            thread2.start();
        }
    }

    static class Local1 implements Runnable {

        @Override
        public void run() {
            ThreadLocalUtils.set("这是local1");
            System.out.println("thread-local1    " + ThreadLocalUtils.get());
        }
    }

    static class local2 implements Runnable {

        @Override
        public void run() {
            ThreadLocalUtils.set("这是local2");
            System.out.println("thread-local2    " + ThreadLocalUtils.get());
        }
    }
}
public class ThreadLocalUtils {
    private static final ThreadLocal<String> local = new ThreadLocal<>();

    public static void set(String value) {
        local.set(value);
    }

    public static String get() {
        return local.get();
    }
}

ThreadLocal保存了当前线程的副本，多个线程之间互不干扰，我们看一下ThreadLocal内部是如何实现的。
1.set()方法

   public void set(T value) {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null)
            map.set(this, value);
        else
            createMap(t, value);
    }

    ThreadLocalMap getMap(Thread t) {
        return t.threadLocals;
    }

ThreadLocalMap 是thread的一个成员变量，是和thread相关联的。

2.TreadLocal是懒加载的。第一次设置的时候，如果map为null才会创建，代码如下：

     ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
            table = new Entry[INITIAL_CAPACITY];
            int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
            table[i] = new Entry(firstKey, firstValue);
            size = 1;
            setThreshold(INITIAL_CAPACITY);
        }

由此可见，内部创建了一个entry数组，相关的副本存放在数组里，且size=1。
3.entry为ThreadLocalMap的内部类

    //null的键值(entry.get()==null)意味着key不再被引用了，所以entry可以被移除
    static class Entry extends WeakReference<ThreadLocal<?>> {
            /** The value associated with this ThreadLocal. */
            Object value;

            Entry(ThreadLocal<?> k, Object v) {
                super(k);
                value = v;
            }
        }

5.结束

本文参考《Java 并发编程的艺术》