并发编程（一）
2022-09-22 22:47:33 #JUC #多线程

hopeStation/thread-learn

并发编程

基础概念

JUC

它是java.util .concurrent工具包的简称。这是一个处理线程的工具包，JDK 1.5开始出现的

进程

指在系统中正在运行的一个应用程序；程序一旦运行就是进程；是系统进行资源分配和调度的基本单位

线程

是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。Java里默认有两个线程：主线程（Main方法）、和垃圾回收（GC）线程

创建线程的方式有三种：

  • 继承Thread
  • 实现Runnable接口
  • 实现Callable接口

在调用start方法开启线程时，底层调用start0方法，接下来start0()方法会调用JVM_StartThread()方法，是通过C++或C的代码，来通过操作系统创建线程的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public synchronized void start() {
    if (this.threadStatus != 0) {
        throw new IllegalThreadStateException();
    } else {
        this.group.add(this);
        boolean var1 = false;

        try {
            //调用start0
            this.start0();
            var1 = true;
        } finally {
            try {
                if (!var1) {
                    this.group.threadStartFailed(this);
                }
            } catch (Throwable var8) {
            }

        }

    }
}
//本地方法
private native void start0();

并发、并行

  • 并发：Concurrent
  • 并行：Parallel

Erlang 之父 Joe Armstrong 用一张5岁小孩都能看懂的图解释了并发与并行的区别

img

并发是两个队列交替使用一台咖啡机，并行是两个队列同时使用两台咖啡机，如果串行，一个队列使用一台咖啡机，那么哪怕前面那个人便秘了去厕所呆半天，后面的人也只能死等着他回来才能去接咖啡，这效率无疑是最低的。

并发和并行都可以是很多个线程，就看这些线程能不能同时被（多个）cpu执行，如果可以就说明是并行，而并发是多个线程被（一个）cpu 轮流切换着执行。

CPU核心数

java.lang.Runtime.availableProcessors() 方法: 返回可用处理器的Java虚拟机的数量

这个值可以在虚拟机中的某个调用过程中改变。应用程序是可用的处理器数量敏感，因此应该偶尔查询该属性，并适当调整自己的资源使用情况。

1
2
3
4
5
public class Demo {
    public static void main(String[] args) {
        System.out.println("" + Runtime.getRuntime().availableProcessors());
    }
}

并发编程的本质：充分利用CPU的资源

Sleep、Wait

区别如下

  • 类：这两个方法来自不同的类分别是Thread和Object

  • 是否释放锁：最主要是sleep方法没有释放同步锁，而wait方法释放了同步锁，使得其他线程可以使用同步控制块或者方法。

  • 使用范围 ：wait，notify和notifyAll只能在同步控制方法或者同步控制块里面使用，而sleep可以在任何地方使用（使用范围）

  • 作用：sleep是Thread类的静态方法。sleep的作用是让线程休眠制定的时间，在时间到达时恢复

    wait是Object的方法，也就是说可以对任意一个对象调用wait方法，调用wait方法将会将调用者的线程挂起，直到其他线程调用同一个对象的notify方法才会重新激活调用者。

而实际上常用的睡眠方法是java.util.concurrent.TimeUnit包下的方法

1
2
3
4
5
6
7
8
9
 public static void main(String[] args) throws InterruptedException {
        System.out.println(new Date());
        TimeUnit.SECONDS.sleep(10);//等待10秒
        System.out.println(new Date());
        //TimeUnit.DAYS.sleep(1);//等待1天，还有小时，分钟等单位
}
//输出：
Thu Nov 12 11:21:59 GMT+08:00 2020
Thu Nov 12 11:22:09 GMT+08:00 2020

线程的状态

线程的状态，在Java中规定在了Thread.State这个枚举类型中，源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
public enum State {
    /**
     * Thread state for a thread which has not yet started.
     *
     * NEW：一个尚未启动的线程的状态。也称之为初始状态、开始状态。
     */
    //
    NEW,

    /**
     * Thread state for a runnable thread.  A thread in the runnable
     * state is executing in the Java virtual machine but it may
     * be waiting for other resources from the operating system
     * such as processor.
     *
     * RUNNABLE：一个可以运行的线程的状态，可以运行是指这个线程已经在JVM中运行了，
     * 但是有可能正在等待其他的系统资源。也称之为就绪状态、可运行状态。
     */
    RUNNABLE,

    /**
     * Thread state for a thread blocked waiting for a monitor lock.
     * A thread in the blocked state is waiting for a monitor lock
     * to enter a synchronized block/method or
     * reenter a synchronized block/method after calling
     * {@link Object#wait() Object.wait}.
     *
     * BLOCKED：一个线程因为等待监视锁而被阻塞的状态。也称之为阻塞状态。
     */
    BLOCKED,

    /**
     * Thread state for a waiting thread.
     * A thread is in the waiting state due to calling one of the
     * following methods:
     * <ul>
     *   <li>{@link Object#wait() Object.wait} with no timeout</li>
     *   <li>{@link #join() Thread.join} with no timeout</li>
     *   <li>{@link LockSupport#park() LockSupport.park}</li>
     * </ul>
     *
     * <p>A thread in the waiting state is waiting for another thread to
     * perform a particular action.
     *
     * For example, a thread that has called <tt>Object.wait()</tt>
     * on an object is waiting for another thread to call
     * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
     * that object. A thread that has called <tt>Thread.join()</tt>
     * is waiting for a specified thread to terminate.
     *
     * WAITING：一个正在等待的线程的状态。也称之为等待状态。造成线程等待的原因有三种:
     * 分别是调用Object.wait()、join()以及LockSupport.park()方法。
     * 处于等待状态的线程，正在等待其他线程去执行一个特定的操作。
     * 例如：因为wait()而等待的线程正在等待另一个线程去调用notify()或notifyAll();
     * 一个因为join()而等待的线程正在等待另一个线程结束。
     */
    WAITING,

    /**
     * Thread state for a waiting thread with a specified waiting time.
     * A thread is in the timed waiting state due to calling one of
     * the following methods with a specified positive waiting time:
     * <ul>
     *   <li>{@link #sleep Thread.sleep}</li>
     *   <li>{@link Object#wait(long) Object.wait} with timeout</li>
     *   <li>{@link #join(long) Thread.join} with timeout</li>
     *   <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
     *   <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
     * </ul>
     *
     * TIMED_WAITING：一个在限定时间内等待的线程的状态。也称之为限时等待状态。
     * 造成线程限时等待状态的原因有五种:
     * 分别是：Thread.sleep(long)、Object.wait(long)、join(long)、
     * LockSupport.parkNanos(obj,long)和LockSupport.parkUntil(obj,long)。
     *
     */
    TIMED_WAITING,

    /**
     * Thread state for a terminated thread.
     * The thread has completed execution.
     *
     * TERMINATED：一个完全运行完成的线程的状态。也称之为终止状态、结束状态。
     */
    TERMINATED;
}

线程状态转换图

image.png

Lock

Lock是一个接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
package java.util.concurrent.locks;

public interface Lock {
    void lock();

    void lockInterruptibly() throws InterruptedException;

    boolean tryLock();

    boolean tryLock(long var1, TimeUnit var3) throws InterruptedException;

    void unlock();

    Condition newCondition();
}

实现类

image20201112132926316.png

常用的锁：ReentrantLock（可重入锁）

1
2
3
4
5
6
7
8
9
10
11
12
13
public class ReentrantLock implements Lock, Serializable {
    public ReentrantLock() {
        //默认是非公平锁
        this.sync = new ReentrantLock.NonfairSync();
    }

    public ReentrantLock(boolean var1) {
        //根据参数确定
        this.sync = (ReentrantLock.Sync)(var1 ? new ReentrantLock.FairSync() : new ReentrantLock.NonfairSync());
    }
    ......
}

  • NonfairSync 非公平锁：也就是没有先后顺序，看CPU心情（调度顺序），Synchronized就是非公平锁
  • FairSync 公平锁，进行排队，先来的（等待时间久的）先执行，后来的（等待时间短的）后执行

Synchronized与Lock的区别

  1. 来源：
    lock是一个接口，而synchronized是java的一个关键字，synchronized是内置的语言实现

    synchronized和lock的用法区别

    synchronized：在需要同步的对象中加入此控制，synchronized可以加在方法上，也可以加在特定代码块中，括号中表示需要锁的对象。

    lock：一般使用ReentrantLock类做为锁。在加锁和解锁处需要通过lock()和unlock()显示指出。所以一般会在finally块中写unlock()以防死锁。

  2. 类型
    synchronized是可重入锁，不可以中断的，非公平
    Lock可重入锁，可以中断，默认非公平，根据参数可以设置公平、非公平

  3. 异常是否释放锁：
    synchronized在发生异常时候会自动释放占有的锁，因此不会出现死锁；而lock发生异常时候，不会主动释放占有的锁，必须手动unlock来释放锁，可能引起死锁的发生。（所以最好将同步代码块用try catch包起来，finally中写入unlock，避免死锁的发生。）

  4. 是否响应中断
    lock等待锁过程中可以用interrupt来中断等待，而synchronized只能等待锁的释放，不能响应中断

    当锁一直在阻塞的时候，lock锁还可以通过lock.tryLock(2000L,TimeUnit.MILLISECONDS)来指定尝试获取的时间

    if (lock.tryLock(2000L,TimeUnit.MILLISECONDS)) { //获锁成功代码段 }

    else{ //具体获取锁失败的回复响应 }

  5. 是否知道获取锁
    Lock可以通过trylock来知道有没有获取锁，而synchronized不能；

  6. 读写分离
    Lock可以提高多个线程进行读操作的效率。（可以通过readwritelock实现读写分离）

  7. 性能
    如果竞争资源不激烈，两者的性能是差不多的，而当竞争资源非常激烈时（即有大量线程同时竞争），此时Lock的性能要远远优于synchronized。所以说，在具体使用时要根据适当情况选择。

    synchronized和lock性能区别

    synchronized是托管给JVM执行的，
    而lock是java写的控制锁的代码。

    在Java1.5中，synchronize是性能低效的。因为这是一个重量级操作，需要调用操作接口，导致有可能加锁消耗的系统时间比加锁以外的操作还多。相比之下使用Java提供的Lock对象，性能更高一些。

    但是到了Java1.6，发生了变化。synchronize在语义上很清晰，可以进行很多优化，有适应自旋，锁消除，锁粗化，轻量级锁，偏向锁等等。导致在Java1.6上synchronize的性能并不比Lock差。官方也表示，他们也更支持synchronize，在未来的版本中还有优化余地。

    2种机制的具体区别：
    synchronized原始采用的是CPU悲观锁机制，即线程获得的是独占锁。独占锁意味着其他线程只能依靠阻塞来等待线程释放锁。而在CPU转换线程阻塞时会引起线程上下文切换，当有很多线程竞争锁的时候，会引起CPU频繁的上下文切换导致效率很低。

    而Lock用的是乐观锁方式。所谓乐观锁就是，每次不加锁而是假设没有冲突而去完成某项操作，如果因为冲突失败就重试，直到成功为止。乐观锁实现的机制就是CAS操作（Compare and Swap）。我们可以进一步研究ReentrantLock的源代码，会发现其中比较重要的获得锁的一个方法是compareAndSetState。这里其实就是调用的CPU提供的特殊指令。

    现代的CPU提供了指令，可以自动更新共享数据，而且能够检测到其他线程的干扰，而 compareAndSet() 就用这些代替了锁定。这个算法称作非阻塞算法，意思是一个线程的失败或者挂起不应该影响其他线程的失败或挂起的算法。

  8. 调度
    synchronized使用Object对象本身的wait 、notify、notifyAll调度机制，而Lock可以使用Condition进行线程之间的调度

  9. 使用场景
    synchronized时候锁少量的同步代码块，Lock适合锁大量的同步代码块

    synchronized和lock用途区别

    synchronized原语和ReentrantLock在一般情况下没有什么区别，但是在非常复杂的同步应用中，请考虑使用ReentrantLock，特别是遇到下面2种需求的时候。

    1.某个线程在等待一个锁的控制权的这段时间需要中断
    2.需要分开处理一些wait-notify，ReentrantLock里面的Condition应用，能够控制notify哪个线程
    3.具有公平锁功能，每个到来的线程都将排队等候

    下面细细道来……

    先说第一种情况，ReentrantLock的lock机制有2种，忽略中断锁和响应中断锁，这给我们带来了很大的灵活性。比如：如果A、B 2个线程去竞争锁，A线程得到了锁，B线程等待，但是A线程这个时候实在有太多事情要处理，就是一直不返回，B线程可能就会等不及了，想中断自己，不再等待这个锁了，转而处理其他事情。这个时候ReentrantLock就提供了2种机制：可中断/可不中断
    第一，B线程中断自己（或者别的线程中断它），但是ReentrantLock不去响应，继续让B线程等待，你再怎么中断，我全当耳边风（synchronized原语就是如此）；
    第二，B线程中断自己（或者别的线程中断它），ReentrantLock处理了这个中断，并且不再等待这个锁的到来，完全放弃。

生产者和消费者问题

synchronized 版

synchronized + wait + notify 解决

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
package JUC;

public class pc {
    public static void main(String[] args) {
        Data data = new Data();

        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                try {
                    data.incrment();
                    Thread.sleep(101);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                try {
                    data.decrment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                try {
                    data.incrment();
                    Thread.sleep(101);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                try {
                    data.decrment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

/**
 * 资源类
 */
class Data{
    private  int number = 0;

    //+1
    public  synchronized  void incrment() throws InterruptedException {
        //这里不能用if，需要用while
        while(number!=0){
            //等待
            this.wait();
        }
        //业务操作
        number++;
        System.out.println(Thread.currentThread().getName() + "==>" + number);
        //通知其他线程，我已经操作完了
        this.notifyAll();
    }

    //-1
    public  synchronized  void decrment() throws InterruptedException {
        //这里不能用if，需要用while
        while(number==0){
            //等待
            this.wait();
        }
        //业务操作
        number--;
        System.err.println(Thread.currentThread().getName() + "==>" + number);
        //通知其他线程，我已经操作完了
        this.notifyAll();
    }
}

注意 synchronized 里修饰的方法，在判断条件不满的等待时，一定要求while。防止虚假唤醒。这也是官方文档中说明的。还可以参考此博客 https://www.cnblogs.com/audi-car/p/6063567.html

image20201118215222659.png

Lock版

image20201118224209624.png

官方文档例子 https://tool.oschina.net/apidocs/apidoc?api=jdk-zh

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class BoundedBuffer {
   final Lock lock = new ReentrantLock();
   final Condition notFull  = lock.newCondition(); 
   final Condition notEmpty = lock.newCondition(); 

   final Object[] items = new Object[100];
   int putptr, takeptr, count;

   public void put(Object x) throws InterruptedException {
     lock.lock(); try {
       while (count == items.length)
         notFull.await();
       items[putptr] = x;
       if (++putptr == items.length) putptr = 0;
       ++count;
       notEmpty.signal();
     } finally { lock.unlock(); }
   }

   public Object take() throws InterruptedException {
     lock.lock(); try {
       while (count == 0)
         notEmpty.await();
       Object x = items[takeptr];
       if (++takeptr == items.length) takeptr = 0;
       --count;
       notFull.signal();
       return x;
     } finally { lock.unlock(); }
   }
 }

生产者消费者例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
package JUC;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class PC_lock {
    public static void main(String[] args) {
        Data1 data = new Data1();

        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                try {
                    data.incrment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                try {
                    data.decrment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                try {
                    data.incrment();
                    Thread.sleep(101);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                try {
                    data.decrment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

/**
 * 资源类
 */
class Data1{
    private  int number = 0;
    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();

    //+1
    public   void incrment() throws InterruptedException {
        lock.lock();
        try {
            while(number!=0){
                //等待
                condition.await();
            }
            //业务操作
            number++;
            System.out.println(Thread.currentThread().getName() + "==>" + number);
            //通知其他线程，我已经操作完了
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    //-1
    public   void decrment() throws InterruptedException {
        lock.lock();
        try {
            while(number==0){
                //等待
                condition.await();
            }
            //业务操作
            number--;
            System.out.println(Thread.currentThread().getName() + "==>" + number);
            //通知其他线程，我已经操作完了
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
}

多线程顺序执行

Condition 的精准通知

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package JUC;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 设计多个线程顺序执行 A-》B-》C
 */
public class PC_lock02 {
    public static void main(String[] args) {
        Data2 data = new Data2();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                data.printC();
            }
        },"C").start();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                data.printA();
            }
        },"A").start();
        new Thread(()->{
            for (int i = 0; i < 20; i++) {
                data.printB();
            }
        },"B").start();
    }
}

/**
 * 资源类
 */
class Data2{
    private Lock lock = new ReentrantLock();
    Condition condition1 = lock.newCondition();
    Condition condition2 = lock.newCondition();
    Condition condition3 = lock.newCondition();
    private  int number = 1; //1A,2B,3C
    public void printA(){
        lock.lock();

        try {
            //业务代码
            //一般顺序都为：判断=》执行=》通知
            while (number!=1){
                //等待
                condition1.await();
            }
            System.out.println(Thread.currentThread().getName() + "==> AAA");
            number = 2;
            //这是故意只唤醒 2 来达到顺序执行的目的
            condition2.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public  void  printB(){
        lock.lock();
        try {
            while (number!=2){
                condition2.await();
            }
            System.out.println(Thread.currentThread().getName() + "==> BBBBBB");
            number=3;
            condition3.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public  void  printC(){
        lock.lock();
        try {
            while (number!=3){
                condition3.await();
            }
            System.out.println(Thread.currentThread().getName() + "==> CCCCCCCCCCC");
            number=1;
            condition1.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

synchronized的原理

待补充…

参考资料

并发与并行的区别是什么？ - 刘志军的回答 - 知乎 https://www.zhihu.com/question/33515481/answer/199929767

hanchao5272 的 线程状态：https://blog.csdn.net/hanchao5272/article/details/79533700
洞玄之境 的 synchronized 和Lock区别:https://blog.csdn.net/hefenglian/article/details/82383569

Prev
2022-09-22 22:47:33 #JUC #多线程
Next