多线程

2022-02-18 本文已影响0人 ttyttytty

线程安全性

多线程时，使用同步机制，对于可变的共享对象的访问/修改，需要保证数据的正常。- 注：访问也是需要同步的（参考内存可见性）。
线程的局部变量在独立的线程栈上，是线程安全的。

线程原则

原子性

多个关联的原子变量/代码块之间，也需要确保原子性修改。

原子类AtomicXX

加内置锁

内存可见性

读为什么也需要同步的原因：保证线程间的修改即时立刻可见。（与写，同一个同步实现）
指令的重排列，无法保证执行的时序。
锁前后，确保工作内存&主存的及时刷新。

volatile-不重排

仅确保修改后，其他线程，立刻可见；不能保证修改时原子的。锁可以可见性+原子性。
当旦仅当满足以下所有条件时，才应该使用 volatile 变量：
1.对变量的写人操作不依赖变量的当前值，或者你能确保只有单个线程更新变量的值。
2.该变量不会与其他状态变量一起纳人不变性条件中。
3.在访问变量时不需要加锁

锁

不变性

不可变对象是必然线程安全的。

不安全问题

1.竞态条件(Race Condition)----原子性

并发时，由于不恰当的执行时序，导致不正确的数据。e.g.a++；可变的共享变量+并发，非同步原子
的情况下，根据多线程的调度，数据不可靠。

2.数据竞争（Data Race）

未同步，导致读写线程的数据时不正确的

3.活跃性问题：操作无法正常执行下去。e.g.死锁/饥饿/活锁

4.发布和逸出

发布：使得对象在当前作用域之外被使用。
逸出：当不该发布的对象被发布出去。
- this引用逸出
在对象未构造完成之前，就发布该对象，破坏线程的安全性。
构造函数中，发布其他对象×。super.register(new listener())，内部实例持有对象的隐含引用
构造函数中，构造一个线程√，此时，线程持有当前对象this,启动一个线程×。
构造函数中，调用可被重写的方法（final private）

5.线程封闭

对象封闭在线程中，不被其他共享。

线程栈/局部变量+不逸出

ThreadLocal（全局变量）

同步机制

锁

原子类AtomicXX

volatile

线程安全的数据结构

Vetor

跨线程的回调函数是如何实现的?

重点在于进程间通信，本质上是如何将与线程相关的变量或者对象传递给别的线程，从而实现交互。
对象是存在多线程共享的堆中的，不是局部变量的线程栈中，所以可以获取。

CountDownLatch : 一个线程(或者多个)，等待另外N个线程完成某个事情之后才能执行。
CyclicBarrier : N个线程相互等待，任何一个线程完成之前，所有的线程都必须等待，类似有福同享，有难同当

因为在泛型代码内部，无法获取任何有关泛型参数类型的任何信息！，Java的泛型就是使用擦除来实现的
那如何获取参数类型？——类型标签Class< T>

typetoken获取类型转换？

全研成的单例内存返回（全局，Rxjava怎么会返回到第二次的？）
Rxjava如何实现线程切换，防止线程泄露？

胜澜的命令异步回调多线程内存泄露（全局变量缓存局部变量，且没有清空操作造成内存泄露https://blog.csdn.net/weixin_39629679/article/details/114221004）

多线程各场景及解决方案


/**
 * scenario:写写未互斥：两个线程同时写同一个变量，
 * problem: 写非原子操作，数据异常
 * solution:写写互斥
 */
public class MultiWriteV1 {
    private int num = 0;

    private void read() {
        System.out.println(num);
    }

    private void write(int change) {
        num += change;
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
        }).start();

        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
        }).start();

        // 睡眠一秒保证线程执行完成 todo
        // 1s一定OK？
        Thread.sleep(1000);
        // 读取结果
        read();
    }
}


/**
 * scenario:写写互斥：两个线程同时写同一个变量，
 * problem: 写写互斥
 * solution:写线程，对象锁，
 */
public class MultiWriteV2 {
    private int num = 0;
    private final Object lock = new Object();

    private void read() {
        System.out.println(num);
    }

    private void write(int change) {
        synchronized (lock) {
            num += change;
        }
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
        }).start();

        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
        }).start();


        // 睡眠一秒保证线程执行完成 todo
        // 1s一定OK？
        Thread.sleep(1000);
        // 读取结果
        read();
    }
}


/**
 * scenario:边读边写
 * solution:写写互斥，读写不互斥，读读不互斥
 * problem: 读到的不是最新主内存中的数据
 */
public class MultiReadWriteV1 {
    private int num = 0;
    private final Object lock = new Object();

    private void read() {
        System.out.println("read " + num);
    }

    private void write(int change) {
        synchronized (lock) {
            num += change;
            System.out.println("write " + num);
        }
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
        }).start();

        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
        }).start();

        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                read();
            }
            System.out.println("read finish");
        }).start();


    }
}


/**
 * scenario:边读边写
 * solution:写写互斥，读写互斥，读读互斥：读写互斥都加同一把锁
 * problem: 读到的是最新主内存中的数据，满足边读边写，但效率低，所有操作都互斥
 */
public class MultiReadWriteV2 {
    private int num = 0;
    private final Object lock = new Object();

    private void read() {
        synchronized (lock) {
            //所以我要给 read 中的代码也加上判断，它也要拿到钥匙后才能读取，
            // 这样就能保证读取时不会有写操作，写的时候也没有读取操作了
            //更常见的需求是写入全部完成后，再去读取值。
            System.out.println("read " + num);
        }
    }

    private void write(int change) {
        synchronized (lock) {
            num += change;
            System.out.println("write " + num);
        }
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
        }).start();

        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
        }).start();

        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                read();
            }
            System.out.println("read finish");
        }).start();


    }
}

/**
 * scenario:边读边写
 * solution:写写互斥，读写不互斥，读读不互斥，volatile
 * problem: 无效 读到的不是最新的
 */
public class MultiReadWriteV3 {
    private volatile int num = 0;
    private final Object lock = new Object();

    private void read() {
        System.out.println("read " + num);
    }

    private void write(int change) {
        synchronized (lock) {
            num += change;
            System.out.println("write " + num);
        }
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
        }).start();

        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
        }).start();

        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                read();
            }
            System.out.println("read finish");
        }).start();
    }
    
}

/**
 * scenario:写完再读
 * solution:写写互斥，读写互斥，读读互斥，加一个写线程是否结束的标志位，读线程循环检查标准位
 * problem: 一旦读在写完之前取拿到了锁，死循环在读线程while，一直打印waiting，写线程无法拿到锁，无法修改度的条件标志位；所有都互斥，效率低
 */
public class MultiReadWriteV4 {
    private int num = 0;
    private final Object lock = new Object();
    private boolean isWriteFinished1 = false;
    private boolean isWriteFinished2 = false;

    private void read() {
        synchronized (lock) {
            //所以我要给 read 中的代码也加上判断，它也要拿到钥匙后才能读取，
            // 这样就能保证读取时不会有写操作，写的时候也没有读取操作了
            //更常见的需求是写入全部完成后，再去读取值。

            //这不能实现10000次读取
//            if (isWriteFinished1 && isWriteFinished2) {
//                System.out.println("read " + num);
//            } else {
//                System.out.println("writing");
//            }

            while (!isWriteFinished1 || !isWriteFinished2) {
                System.out.println("writing");
            }
            System.out.println("read " + num);
        }
    }

    private void write(int change) {
        synchronized (lock) {
            num += change;
            System.out.println("write " + num);
        }
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
            isWriteFinished1 = true;
        }).start();
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
            isWriteFinished2 = true;
        }).start();


        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                read();
            }
            System.out.println("read finish");
        }).start();
    }
}

/**
 * scenario:写完再读
 * solution:写写互斥，读写互斥，读读互斥，不阻塞，等待/唤醒机制（synchronized (lock) {lock.wait、lock.notify}））
 * 写入操作不受限制；如果写入还没有完成，read 方法先进入等待状态。write 方法写入完成后，通知 read 开始读取
 * problem: 读读互斥，效率低
 */
public class MultiReadWriteV5 {
    private int num = 0;
    private final Object lock = new Object();
    private boolean isWriteFinished1 = false;
    private boolean isWriteFinished2 = false;

    private void read() {
        synchronized (lock) {
            while (!isWriteFinished1 || !isWriteFinished2) {
                // 等待，并且不要阻塞写入
                try {
                    lock.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 只会在上述的lock被notify，往下走后，打印两次
                System.out.println("writing");
            }
            System.out.println("read " + num);
        }
    }

    private void write(int change) {
        synchronized (lock) {
            num += change;
            System.out.println("write " + num);
        }
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
            isWriteFinished1 = true;
            //写线程写完后，唤醒读取线程继续读取。
            synchronized (lock) {
                lock.notify();
            }
        }).start();
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
            isWriteFinished2 = true;
            //写线程写完后，唤醒读取线程继续读取。
            synchronized (lock) {
                lock.notify();
            }
        }).start();


        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                read();
            }
            System.out.println("read finish");
        }).start();
    }
}

/**
 * scenario:写完再读
 * solution:写写互斥，读写互斥，不阻塞，等待/唤醒机制（ReentrantLock+Condition 类替代synchronized (lock) {lock.wait、lock.notify}））
 * ReentrantLock可以设置尝试获取锁的等待时间+condition可以设置自唤醒时间
 * 写入操作不受限制；如果写入还没有完成，read 方法先进入等待状态。write 方法写入完成后，通知 read 开始读取
 * problem:读读互斥，效率低
 */
public class MultiReadWriteV6 {
    private int num = 0;

    private final ReentrantLock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();

    private boolean isWriteFinished1 = false;
    private boolean isWriteFinished2 = false;

    private void read() {
        lock.lock();

        while (!isWriteFinished1 || !isWriteFinished2) {
            // 等待，并且不要阻塞写入
            try {
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 只会在上述的lock被notify，往下走后，打印两次
            System.out.println("writing");

//            if (condition.await(1, TimeUnit.SECOND)) {
//                // 1 秒内被 signal 唤醒
//            } else {
//                // 1 秒内没有被唤醒，自己醒来
//            }


        }
        System.out.println("read " + num);

        lock.unlock();
    }

    private void write(int change) {
        try {
            if (lock.tryLock(1, TimeUnit.SECONDS)) {
                num += change;
                System.out.println("write " + num);

                lock.unlock();
            } else {
                System.out.println("1 秒内没有获取到锁，不再等待。不执行");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
            isWriteFinished1 = true;

            //写线程写完后，唤醒读取线程继续读取。
            // 写入完成，唤醒读取线程，await/signal 操作必须在 lock 时执行。
            lock.lock();
            condition.signal();
            lock.unlock();

        }).start();
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
            isWriteFinished2 = true;
            //写线程写完后，唤醒读取线程继续读取。
            // 写入完成，唤醒读取线程，await/signal 操作必须在 lock 时执行。
            lock.lock();
            condition.signal();
            lock.unlock();
        }).start();


        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                read();
            }
            System.out.println("read finish");
        }).start();
    }
}

/**
 * scenario:写完再读
 * solution:写写互斥，读写互斥，读读互斥（ReadWriteLock优化读读互斥，本身也支持不阻塞的唤醒通知机制）
 * problem:ReadWriteLock 会导致写线程必须等待读线程完成后才能写（悲观锁）
 */
public class MultiReadWriteV7 {
    private int num = 0;

    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final Lock writeLock = lock.writeLock();
    private final Lock readLock = lock.readLock();
    private final Condition condition = writeLock.newCondition();

    private boolean isWriteFinished1 = false;
    private boolean isWriteFinished2 = false;

    private void read() {
        readLock.lock();

        while (!isWriteFinished1 || !isWriteFinished2) {
            // 等待，并且不要阻塞写入
            try {
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 只会在上述的lock被notify，往下走后，打印两次
            System.out.println("writing");

//            if (condition.await(1, TimeUnit.SECOND)) {
//                // 1 秒内被 signal 唤醒
//            } else {
//                // 1 秒内没有被唤醒，自己醒来
//            }


        }
        System.out.println("read " + num);

        readLock.unlock();
    }

    private void write(int change) {
        try {
            if (writeLock.tryLock(1, TimeUnit.SECONDS)) {
                num += change;
                System.out.println("write " + num);

                writeLock.unlock();
            } else {
                System.out.println("1 秒内没有获取到锁，不再等待。不执行");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
            isWriteFinished1 = true;

            //写线程写完后，唤醒读取线程继续读取。
            // 写入完成，唤醒读取线程，await/signal 操作必须在 lock 时执行。
            writeLock.lock();
            condition.signal();
            writeLock.unlock();

        }).start();
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
            isWriteFinished2 = true;
            //写线程写完后，唤醒读取线程继续读取。
            // 写入完成，唤醒读取线程，await/signal 操作必须在 lock 时执行。
            writeLock.lock();
            condition.signal();
            writeLock.unlock();
        }).start();


        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                read();
            }
            System.out.println("read finish");
        }).start();
    }
}

/**
 * scenario:写完再读
 * solution:写写互斥，读写不互斥（ StampedLock乐观锁，读的过程中也允许写。通过版本对比判断读的过程中是否有写入发生+回退悲观锁的方案，确保能读到主内存中最新的值）
 * problem:
 */
public class MultiReadWriteV8 {
    private int num = 0;
    private final StampedLock lock = new StampedLock();

    private void read() {
        long stamp = lock.tryOptimisticRead();
        int readNumber = num;
        if (!lock.validate(stamp)) {
            stamp = lock.readLock();
            System.out.println("乐观读取到的 number " + readNumber + " 有误，换用悲观锁重新读取：number = " + num);
            lock.unlockRead(stamp);
        }
    }

    private void write(int change) {
        long stamp = lock.writeLock();
        num += change;
        System.out.println("write " + num);
        lock.unlockWrite(stamp);
    }

    public void readAndWrite() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(1);
            }
            System.out.println("+ 10000 finish");
        }).start();
        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                write(-1);
            }
            System.out.println("- 10000 finish");
        }).start();


        new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                read();
            }
            System.out.println("read finish");
        }).start();
    }
}

锁

ReentrantLock
公平锁&非公平锁：抢锁直接排队还是先去抢失败再排队，ReentrantLock默认非公平
偏向锁、轻量锁等
乐观锁（无锁算法CAS，先干起来，写的时候检查是否被其他线程修改，修改的话就；AtomicInteger Java原子类中的递增操作就通过Compare And Swap（比较与交换自旋实现）&悲观锁（锁多写场景，，synchronized关键字和Lock的实现类）：锁还是不锁同步资源
自旋锁&适应性自旋：抢锁失败了，阻塞这个线程还是让他自旋：多个处理器，CPU切换+恢复线程线程的消耗>自旋*N次的消耗。因为切换线程阻塞挂起唤醒等需要切CPU，保存/恢复现场，消耗大。如果同步资源获取后，线程处理逻辑简单，其实是比阻塞线程开销少。如果有多个处理器，可以让后来的线程自及忙一会，等会再来检查同步资源，到时候直接获取，避免线程切换。
可重入锁：同一线程能否多次获取同一把锁，ReentrantLock和synchronized都是重入锁
共享锁(大家都可读)&排他/独享锁（我自己可以读可以写）：多个线程能否共享一把锁？那还锁啥锁？
https://tech.meituan.com/2018/11/15/java-lock.html
https://segmentfault.com/a/1190000023735772

多线程

线程安全性

线程原则

原子性

原子类AtomicXX

加内置锁

内存可见性

volatile-不重排

锁

不变性

不安全问题

1.竞态条件(Race Condition)----原子性

2.数据竞争（Data Race）

3.活跃性问题：操作无法正常执行下去。e.g.死锁/饥饿/活锁

4.发布和逸出

5.线程封闭

线程栈/局部变量+不逸出

ThreadLocal（全局变量）

同步机制

锁

原子类AtomicXX

volatile

线程安全的数据结构

多线程各场景及解决方案

锁

猜你喜欢

热点阅读