个人学习生活记录小站

JAVA 多线程使用

引用

https://www.liaoxuefeng.com/wiki/1252599548343744/1255943750561472
https://www.cnblogs.com/tong-yuan/p/StampedLock.html
https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html
https://www.cnblogs.com/chenpi/p/5375805.html
https://blog.csdn.net/qq_30379689/article/details/80785650

线程创建

public class Main {
    public static void main(String[] args) {
        Thread t = new MyThread();//通过扩展Thread
        Thread t = new Thread(new MyRunnable());
        Thread t = new Thread(() -> {
            System.out.println("start new thread!");
        });
      
 
        t.start(); // 启动新线程
    }
}

class MyThread extends Thread {
    @Override
    public void run() {
        System.out.println("start new thread!");
    }
}
class MyRunnable implements Runnable {
    @Override
    public void run() {
        System.out.println("start new thread!");
    }
}
  1. 通过扩展Thread创建线程
  2. 通过实现Runable，作为构造函数的参数创建线程
  3. 通过内部类实现Runable，作为构造函数的参数创建线程

线程中断

通过 interrupt()

public class Main {
    public static void main(String[] args) throws InterruptedException {
        Thread t = new MyThread();
        t.start();
        Thread.sleep(1000);
        t.interrupt(); // 中断t线程
        t.join(); // 等待t线程结束
        System.out.println("end");
    }
}

class MyThread extends Thread {
    @Override
    public void run() {
        Thread hello = new HelloThread();
        hello.start(); // 启动hello线程
        try {
            hello.join(); // 等待hello线程结束
        } catch (InterruptedException e) {
            System.out.println("interrupted!");
        }
        hello.interrupt();
    }
}

class HelloThread extends Thread {
    @Override
    public void run() {
        int n = 0;
        while (!isInterrupted()) {
            n++;
            System.out.println(n + " hello!");
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                System.out.println("interrupted!2");
            }
        }
    }
}

//输出
1 hello!
2 hello!
3 hello!
4 hello!
5 hello!
6 hello!
7 hello!
8 hello!
9 hello!
10 hello!
interrupted!
end
interrupted!2
11 hello!
12 hello!
....

我们来详细分析一下这个例子

  1. Main线程开启了MyThread线程，开始睡眠1000ms
  2. MyThread开启了HelloThread,并等待其结束，主要此时有一个InterruptedException异常的捕获
  3. HelloThread每睡眠100ms输出n+hello!，也有一个捕获
  4. Main线程经过1000ms的睡眠，对MyThread线程发出打断指令，并等待其结束
  5. MyThread收到打断指令，停止对HelloThread线程的等待，并抛出InterruptedException异常，被捕获后输出interrupted!
  6. MyThread对HelloThread线程发出打断指令后，MyThread结束后Main先得知并输出end
  7. 此时HelloThread正好处于睡眠，于是抛出异常并输出interrupted!2，但由于打断指令时在睡眠，且异常被捕获了，于是!isInterrupted()仍未true，继续输出`n+hello!

我们从这个例子中可以得知

  1. interrupt()方法仅仅向线程发出了“中断请求”，至于线程是否能立刻响应，要看具体代码。
  2. join的等待和sleep的睡眠，在收到interrupt()打断时都会抛出InterruptedException异常，抛出异常后会影响到isInterrupted()的监测

通过标志位中断

public class Main {
    public static void main(String[] args)  throws InterruptedException {
        HelloThread t = new HelloThread();
        t.start();
        Thread.sleep(1);
        t.running = false; // 标志位置为false
    }
}

class HelloThread extends Thread {
    public volatile boolean running = true;
    public void run() {
        int n = 0;
        while (running) {
            n ++;
            System.out.println(n + " hello!");
        }
        System.out.println("end!");
    }
}
  • 通过标准位中断必须使用volatile
    因为在Java虚拟机中，变量的值保存在主内存中，但是，当线程访问变量时，它会先获取一个副本，并保存在自己的工作内存中。如果线程修改了变量的值，虚拟机会在某个时刻把修改后的值回写到主内存，但是，这个时间是不确定的！

volatile

  1. 每次访问变量时，总是获取主内存的最新值；
  2. 每次修改变量后，立刻回写到主内存。

守护线程

Thread t = new MyThread();
t.setDaemon(true);

JVM在所有非守护线程退出后才会退出，所有非守护线程都执行完毕后，无论有没有守护线程，虚拟机都会自动退出。

可以得知所以

  1. 守护线程是为其他线程服务的线程
  2. 守护线程不能持有任何需要关闭的资源，例如打开文件等，因为虚拟机退出时，守护线程没有任何机会来关闭文件，这会导致数据丢失

线程同步

JAVA通过synchronized对线程进行同步

  1. 修饰实例方法，作用于当前实例加锁，进入同步代码前要获得当前实例的锁
public void add(int n) {
    synchronized(this) { // 锁住实例
        count += n;
    } // 解锁
}
//or
public synchronized void add(int n) { // 锁住实例
    count += n;
} // 解锁
  1. 修饰静态方法，作用于当前类对象加锁，进入同步代码前要获得当前类对象的锁
public class Counter {
    public synchronized static void test(int n) {//锁住Counter类的class实例
            ...
    }// 解锁
}
//or
public class Counter {
    public static void test(int n) {
        synchronized(Counter.class) {//锁住Counter类的class实例
            ...
        }
    }// 解锁
}

synchronized是可重入锁

  1. JVM允许同一个线程重复获取同一个锁，这种能被同一个线程反复获取的锁，就叫做可重入锁。
  2. 每获取一次锁，记录+1，每退出synchronized块，记录-1，减到0的时候，才会真正释放锁。

常用类

wait() 等待

class TaskQueue {
    Queue<String> queue = new LinkedList<>();

    public synchronized void addTask(String s) {
        this.queue.add(s);
        this.notifyAll();
    }

    public synchronized String getTask() throws InterruptedException {
        while (queue.isEmpty()) {
            this.wait();
        }
        return queue.remove();
    }
}
  • 定义在Object类的一个native方法
  • 必须在synchronized块中才能调用wait()方法
  • wait()方法调用时，会释放线程获得的锁，wait()方法返回后，线程又会重新试图获得锁。

notify()、notifyAll() 唤醒

  • 定义在Object类的一个native方法
  • 随机唤醒一个(全部)正在锁等待的线程
  • 唤醒的线程仍需要正常获取到锁才能运行

ReentrantLock 重入锁(Lock实现)

public class Counter {
    private final Lock lock = new ReentrantLock();
    private int count;

    public void add(int n) {
        lock.lock();
//lock.tryLock(1, TimeUnit.SECONDS))
        try {
            count += n;
        } finally {
            lock.unlock();
        }
    }
}
  • ReentrantLock实现了Lock接口
  • ReentrantLock可以替代synchronized进行同步
  • 可以使用tryLock()尝试获取锁(例子中尝试1秒时间内获取锁)

Condition 状态（ReentrantLock中的wait,notify等）

class TaskQueue {
    private final Lock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();
    private Queue<String> queue = new LinkedList<>();

    public void addTask(String s) {
        lock.lock();
        try {
            queue.add(s);
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public String getTask() {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                condition.await();
            }
            return queue.remove();
        } finally {
            lock.unlock();
        }
    }
}
  • Condition实例通常由Lock接口的实例创建
  • await()会释放当前锁，进入等待状态；
  • signal()会唤醒某个等待线程；
  • signalAll()会唤醒所有等待线程；

ReentrantReadWriteLock 重入读写锁(ReadWriteLock实现)

public class Counter {
    private final ReadWriteLock rwlock = new ReentrantReadWriteLock();
    private final Lock rlock = rwlock.readLock();
    private final Lock wlock = rwlock.writeLock();
    private int[] counts = new int[10];

    public void inc(int index) {
        wlock.lock(); // 加写锁
        try {
            counts[index] += 1;
        } finally {
            wlock.unlock(); // 释放写锁
        }
    }

    public int[] get() {
        rlock.lock(); // 加读锁
        try {
            return Arrays.copyOf(counts, counts.length);
        } finally {
            rlock.unlock(); // 释放读锁
        }
    }
}
  • ReentrantReadWriteLock实现了ReadWriteLock
public interface ReadWriteLock {
    Lock readLock();

    Lock writeLock();
}
  • 只允许一个线程写入（其他线程既不能写入也不能读取）
  • 没有写入时，多个线程允许同时读（提高性能）

StampedLock 冲压锁？

public class Point {
    private final StampedLock stampedLock = new StampedLock();

    private double x;
    private double y;

    public void move(double deltaX, double deltaY) {
        long stamp = stampedLock.writeLock(); // 获取写锁
        try {
            x += deltaX;
            y += deltaY;
        } finally {
            stampedLock.unlockWrite(stamp); // 释放写锁
        }
    }

    public double distanceFromOrigin() {
        long stamp = stampedLock.tryOptimisticRead(); // 获得一个乐观读锁
        // 注意下面两行代码不是原子操作
        // 假设x,y = (100,200)
        double currentX = x;
        // 此处已读取到x=100，但x,y可能被写线程修改为(300,400)
        double currentY = y;
        // 此处已读取到y，如果没有写入，读取是正确的(100,200)
        // 如果有写入，读取是错误的(100,400)
        if (!stampedLock.validate(stamp)) { // 检查乐观读锁后是否有其他写锁发生
            stamp = stampedLock.readLock(); // 获取一个悲观读锁
            try {
                currentX = x;
                currentY = y;
            } finally {
                stampedLock.unlockRead(stamp); // 释放悲观读锁
            }
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }
}
  • StampedLock也是一种读写锁，但不是基于AQS实现的（AQS实现的：Lock、Semaphore、ReentrantLock等）
  • StampedLock把读锁细分为乐观读和悲观读，还提供了更复杂的将悲观读锁升级为写锁的功能
  • StampedLock的state的高24位存储的是版本号，写锁的释放会增加其版本号，读锁不会
  • StampedLock的低7位存储的读锁被获取的次数，第8位存储的是写锁被获取的次数；
  • StampedLock不是可重入锁
  • StampedLock不能实现条件锁；

JAVA 并发集合包 concurrent

image.png

  • 使用与集合相同，加锁在方法内部实现

JAVA 原子操作包 concurrent.atomic

image.png

  • 原子操作实现了无锁的线程安全
  • Atomic包里的类基本都是使用Unsafe实现的包装类，核心操作是CAS(Compare and Set)原子操作.现代CPU已广泛支持CAS指令，如果不支持，那么JVM将使用自旋锁，与互斥锁一样，两者都需先获取锁才能访问共享资源，但互斥锁会导致线程进入睡眠，而自旋锁会一直循环等待直到获取锁
  • 系统在硬件层面保证了CAS操作的原子性，不会锁住当前线程，它的效率是很高的。但是在并发越高的条件下，失败的次数会越多，CAS如果长时间不成功，会极大的增加CPU的开销，因此CAS不适合竞争十分频繁的场景

自旋锁

  • 阻塞或唤醒一个Java线程需要操作系统切换CPU状态来完成，这种状态转换需要耗费处理器时间。状态转换消耗的时间有可能比用户代码执行的时间还要长。

  • 当前线程不放弃CPU的执行时间，进行自旋，如果在自旋完成后前面锁定同步资源的线程已经释放了锁，那么当前线程就可以不必阻塞而是直接获取同步资源，从而避免切换线程的开销。这就是自旋锁。

  • 自旋等待虽然避免了线程切换的开销，但它要占用处理器时间，所以，自旋等待的时间必须要有一定的限度，如果自旋超过了限定次数（默认是10次）没有成功获得锁，就应当挂起线程。

至虚极，守静笃

© 2024 个人学习生活记录小站