Java 多线程

基本概念

并发：同一时间对应对（dealing with）多个事件的能力

并行：同一时间去动手处理（doing）多个事件的能力

查看进程线程的方式

Windows

任务管理器
tasklist 查看进程
taskkill 杀死进程

Linux

ps -fe 查看所有进程
ps -fe -p <PID> 查看某个进程(PID)的所有线程
kill 杀死进程
top 按大写H切换是否显示进程
top -H -p <PID> 查看某个进程(PID)的所有线程

Java

jps 查看所有Java进程
jstack <PID> 查看某个Java进程(PID)的所有线程状态
jconsole 查看某个Java进程中线程的运行情况(图形界面)

创建新线程

一、从`Thread`派生一个自定义类，然后重写`run()`方法：

java

package org.example;

public class Main {
  public static void main(String[] args) {
    Thread t = new MyThread();
    t.start(); // 启n动新线程
  }
}

class MyThread extends Thread {
  @Override
  public void run() {
    System.out.println("start new thread!");
  }
}

image-20220720a3317

二、创建`Thread`实例时，传入一个`Runnable`实例：

java

package org.example;

public class Main {
  public static void main(String[] args) {
    Thread t = new Thread(new MyRunnable());
    t.start(); // 启动新线程
  }
}

class MyRunnable implements Runnable {
  @Override
  public void run() {
    System.out.println("start new thread!");
  }
}

三、使用lambda

java

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

线程状态

java

public enum State {
  //线程刚创建
  NEW, //在JVM中正在运行的线程
  RUNNABLE, //线程处于阻塞状态，等待监视锁，可以重新进行同步代码块中执行
  BLOCKED, //等待状态
  WAITING, //调用sleep() join() wait()方法可能导致线程处于等待状态
  TIMED_WAITING, //线程执行完毕，已经退出
  TERMINATED;
}

New：新创建的线程，尚未执行；
Runnable：运行中的线程，正在执行run()方法的Java代码；
Blocked：运行中的线程，因为某些操作被阻塞而挂起；
Waiting：运行中的线程，因为某些操作在等待中；
Timed Waiting：运行中的线程，因为执行sleep()方法正在计时等待；
Terminated：线程已终止，因为run()方法执行完毕。

Waiting状态

join()方法，需要让线程按照指定的顺序执行时使用。

Thread.join()方法表示调用此方法的线程被阻塞，仅当该方法完成以后，才能继续运行。

java

import java.util.concurrent.atomic.AtomicInteger;

public class Main {
  public static void main(String[] args) throws InterruptedException {
    AtomicInteger count = new AtomicInteger();
    Thread t1 = new Thread(() -> {
      System.out.println("Thread t1 run");
      count.set(10);
    });
    t1.start();
    t1.join();//若把这一行注释掉，则打印的count可能为0
    System.out.println("count=" + count);
  }
}

Interrupted状态

interrupt()方法可以中断Runnable状态的线程，或者使Waiting状态的线程抛出异常并结束运行

java

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

class MyThread extends Thread {
  public void run() {
    int n = 0;
    while (!isInterrupted()) {
      n++;
      System.out.println(n + " hello!");
    }
  }
}

通过修改running标志位来中断线程

java

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 关键字修饰的变量可以使线程每次访问变量时，总是获取主内存的最新值，并且在每次修改变量后，立刻回写到主内存。

在x86架构下，JVM回写主内存速度很快，加不加volatile影响不大，若是ARM架构，就会有明显的延迟。

守护线程

守护线程是为其他线程服务的线程，如定时触发的任务；

所有非守护线程都执行完毕后，虚拟机退出；

守护线程不能持有需要关闭的资源（如打开文件等）。

线程同步

synchronized()方法

java

synchronized(Counter.lock){
  // 获取锁
  ...
  }
// 释放锁

java

public class Main {
  public static void main(String[] args) throws Exception {
    Thread[] ts = new Thread[]{new Thread1(), new Thread2(), new Thread3(), new Thread4()};
    for (Thread t : ts) {
      t.start();
    }
    for (Thread t : ts) {
      t.join();
    }
    System.out.println(Counter.count1);
    System.out.println(Counter.count2);
  }
}

class Counter {
  public static final Object lock1 = new Object();
  public static final Object lock2 = new Object();
  public static int count1 = 0;
  public static int count2 = 0;
}

class Thread1 extends Thread {
  public void run() {
    for (int i = 0; i < 10000; i++) {
      synchronized (Counter.lock1) {
        Counter.count1 += 1;
      }
    }
  }
}

class Thread2 extends Thread {
  public void run() {
    for (int i = 0; i < 10000; i++) {
      synchronized (Counter.lock1) {
        Counter.count1 -= 1;
      }
    }
  }
}

class Thread3 extends Thread {
  public void run() {
    for (int i = 0; i < 10000; i++) {
      synchronized (Counter.lock2) {
        Counter.count2 += 1;
      }
    }
  }
}

class Thread4 extends Thread {
  public void run() {
    for (int i = 0; i < 10000; i++) {
      synchronized (Counter.lock2) {
        Counter.count2 -= 1;
      }
    }
  }
}

线程安全

synchronized方法加锁对象是this，如以下代码即为封装后的Counter类，它是一个“线程安全”的类

java

public class Counter {
  private int count = 0;

  public void add(int n) {
    synchronized (this) {
      count += n;
    }
  }

  public void dec(int n) {
    synchronized (this) {
      count -= n;
    }
  }

  public int get() {
    return count;
  }
}

多线程常用辅助类

CountDownLatch

java

public class CountDownLatchDemo {
  public static void main(String[] args) throws InterruptedException {
    // 计数器
    CountDownLatch countDownLatch = new CountDownLatch(6);
    for (int i = 1; i <= 6; i++) {
      new Thread(() -> {
        System.out.println(Thread.currentThread().getName() + " Start");
        countDownLatch.countDown(); // 计数器-1
      }, String.valueOf(i)).start();
    }
    //阻塞等待计数器归零
    countDownLatch.await();
    System.out.println(Thread.currentThread().getName() + " End");
  }

  /**
   * 顺序不一定，结果诡异，达不到预期的最后End
   */
  public void test1() {
    for (int i = 1; i <= 6; i++) {
      new Thread(() -> {
        System.out.println(Thread.currentThread().getName() + " Start");
      }, String.valueOf(i)).start();
    }
    System.out.println(Thread.currentThread().getName() + " End");
  }
}

CyclicBarrier

Semaphore

java

acquire（获取）

当一个线程调用 acquire 操作时，他要么通过成功获取信号量（信号量-1）要么一直等下去，直到有线程释放信号量，或超时

release （释放）实际上会将信号量的值 + 1，然后唤醒等待的线程。

应用场景：多个共享资源的互斥使用；并发线程数的控制。

`wait()` 与 `notify()`

java

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.Queue;

public class Main {
  public static void main(String[] args) throws InterruptedException {
    TaskQueue q = new TaskQueue();
    ArrayList<Thread> ts = new ArrayList<Thread>();
    for (int i = 0; i < 5; i++) {
      Thread t = new Thread() {
        public void run() {
          // 执行task:
          while (true) {
            try {
              String s = q.getTask();
              System.out.println("execute task: " + s);
            } catch (InterruptedException e) {
              return;
            }
          }
        }
      };
      t.start();
      ts.add(t);
    }
    Thread add = new Thread(() -> {
      for (int i = 0; i < 10; i++) {
        // 放入task:
        String s = "t-" + Math.random();
        System.out.println("add task: " + s);
        q.addTask(s);
        try {
          Thread.sleep(100);
        } catch (InterruptedException e) {
          throw new RuntimeException(e);
        }
      }
    });
    add.start();
    add.join();
    Thread.sleep(100);
    for (Thread t : ts) {
      t.interrupt();
    }
  }
}

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();
  }
}

若getTask方法中不适用wait，会一直占用this锁，导致进入死循环，循环中添加对wait 方法的执行后，线程会进入等待状态，直到notify将其唤醒，才会进行下次循环并再次判断队列是否为空。

addTask()方法，内部调用了this.notifyAll()而不是this.notify()，使用notifyAll()将唤醒所有当前正在this 锁等待的线程，而notify()只会唤醒其中一个。这是因为可能有多个线程正在getTask()方法内部的wait() 中等待，使用notifyAll()将一次性全部唤醒。通常来说，notifyAll()更安全

`ReentrantLock`

使用synchronized

java

public class Counter {
  private int count;

  public void add(int n) {
    synchronized (this) {
      count += n;
    }
  }
}

使用ReentrantLock

java

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

  public void add(int n) {
    lock.lock();
    try {
      count += n;
    } finally {
      lock.unlock();
    }
  }
}

使用ReentrantLock尝试获取锁

java

if(lock.tryLock(1,TimeUnit.SECONDS)){
  try{
  ...
  }finally{
  lock.unlock();
  }
  }

使用ReentrantLock可以避免因为使用synchronized时抛出异常导致的死锁。

`Condition`

使用ReentrantLock时，wait与notify方法应使用Condition对象实现

java

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();
    }
  }
}

`ReadWriteLock`

使用ReadWriteLock可以实现同一时间只允许一个线程写入，而在没有写入时，允许多个线程同时读，从而提高性能。

java

ublic

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(); // 释放读锁
    }
  }
}

`StampedLock`（乐观锁）

乐观锁会乐观地估计读的过程中大概率不会有写入，悲观锁则是读的过程中拒绝有写入，也就是写入必须等待。

java

import java.util.concurrent.locks.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);
  }
}

`Concurrent`

接口	非线程安全类	线程安全类
List	`ArrayList`	`CopyOnWriteArrayList`
Map	`HashMap`	`ConcurrentHashMap`
Set	`HashSet` / `TreeSet`	`CopyOnWriteArraySet`
Queue	`ArrayDeque` / `LinkedList`	`ArrayBlockingQueue` / `LinkedBlockingQueue`
Deque	`ArrayDeque` / `LinkedList`	`LinkedBlockingDeque`

示例

Map<String, String> map = new ConcurrentHashMap<>();
// 在不同的线程读写:
map.put("A", "1");
map.put("B", "2");
map.get("A", "1");

`Atomic`

以AtomicInteger为例，它提供的主要操作有：

增加值并返回新值：int addAndGet(int delta)
加1后返回新值：int incrementAndGet()
获取当前值：int get()
用CAS方式设置：int compareAndSet(int expect, int update)

例如使用AtomicLong编写一个多线程安全的全局唯一ID生成器：

java

class IdGenerator {
  AtomicLong var = new AtomicLong(0);

  public long getNextId() {
    return var.incrementAndGet();
  }
}

`ExecutorService`（线程池）

FixedThreadPool：线程数固定的线程池。
CachedThreadPool：线程数根据任务动态调整的线程池。
SingleThreadExecutor：仅单线程执行的线程池。
ScheduledThreadPool：可以放入需要定期反复执行的线程池。

java

import java.util.concurrent.*;

public class Main {
  public static void main(String[] args) {
    ScheduledExecutorService ses = Executors.newScheduledThreadPool(4);
    // 1秒后执行一次性任务:
    ses.schedule(new Task("one-time"), 1, TimeUnit.SECONDS);
    // 2秒后开始执行定时任务，每3秒执行:
    ses.scheduleAtFixedRate(new Task("fixed-rate"), 2, 3, TimeUnit.SECONDS);
    //2秒后开始执行定时任务，以3秒为间隔执行:
    ses.scheduleWithFixedDelay(new Task("fixed-delay"), 2, 3, TimeUnit.SECONDS);
    // 创建一个固定大小的线程池:
    ExecutorService es = Executors.newFixedThreadPool(4);
    for (int i = 0; i < 6; i++) {
      es.submit(new Task("" + i));
    }
    // 关闭线程池:
    es.shutdown();
  }
}

class Task implements Runnable {
  private final String name;

  public Task(String name) {
    this.name = name;
  }

  @Override
  public void run() {
    System.out.println("start task " + name);
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
    }
    System.out.println("end task " + name);
  }
}

为了规避资源耗尽的风险，不使用Executors来创建线程池，使用ThreadPoolExecutor来创建

ThreadPoolExecutor 七大参数

corePoolSize：即使线程是空闲的，也要保留在池中的线程数，除非设置了 allowCoreThreadTimeOut

maximumPoolSize：池中允许的最大线程数

keepAliveTime：当线程数大于核心时，这是多余的空闲线程在终止前等待新任务的最大时间。

unit：keepAliveTime参数的时间单位

workQueue—在执行任务之前用来保存任务的队列。该队列将只保存由execute方法提交的可运行任务。

threadFactory—当执行程序创建新的线程处理程序时使用的工厂

handler—当由于达到线程边界和队列容量而阻塞执行时使用的处理程序

Future

Runnable接口没有返回值，Callable带有返回值。

java

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class Main {
  public static void main(String[] args) throws Exception {
    ExecutorService es = Executors.newFixedThreadPool(4);
    Future<String> future = es.submit(new Task("123"));
    System.out.println(future.get());
    es.shutdown();
  }
}

class Task implements Callable<String> {
  String name;

  public Task(String code) {
    this.name = code;
  }

  @Override
  public String call() throws Exception {
    Thread.sleep(1000);
    return "hello " + this.name;
  }
}

一个Future<V>接口表示一个未来可能会返回的结果，它定义的方法有：

get()：获取结果（任务完成时返回结果，未完成时阻塞，直到完成时才返回结果）
get(long timeout, TimeUnit unit)：获取结果，但只等待指定的时间；
cancel(boolean mayInterruptIfRunning)：取消当前任务；
isDone()：判断任务是否已完成。

`CompletableFuture`

CompletableFuture的优点是：

异步任务结束时，会自动回调某个对象的方法；
异步任务出错时，会自动回调某个对象的方法；
主线程设置好回调后，不再关心异步任务的执行。

import java.util.concurrent.CompletableFuture;

public class Main {
    public static void main(String[] args) throws Exception {
        // 两个CompletableFuture执行异步查询:
        CompletableFuture<String> cfQueryFromSina = CompletableFuture.supplyAsync(() -> {
            return queryCode("中国石油", "https://finance.sina.com.cn/code/");
        });
        CompletableFuture<String> cfQueryFrom163 = CompletableFuture.supplyAsync(() -> {
            return queryCode("中国石油", "https://money.163.com/code/");
        });
        // 用anyOf合并为一个新的CompletableFuture:
        CompletableFuture<Object> cfQuery = CompletableFuture.anyOf(cfQueryFromSina, cfQueryFrom163);
        // 两个CompletableFuture执行异步查询:
        CompletableFuture<Double> cfFetchFromSina = cfQuery.thenApplyAsync((code) -> {
            return fetchPrice((String) code, "https://finance.sina.com.cn/price/");
        });
        CompletableFuture<Double> cfFetchFrom163 = cfQuery.thenApplyAsync((code) -> {
            return fetchPrice((String) code, "https://money.163.com/price/");
        });
        // 用anyOf合并为一个新的CompletableFuture:
        CompletableFuture<Object> cfFetch = CompletableFuture.anyOf(cfFetchFromSina, cfFetchFrom163);
        // 最终结果:
        cfFetch.thenAccept((result) -> {
            System.out.println("price: " + result);
        });
        // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
        Thread.sleep(200);
    }

    static String queryCode(String name, String url) {
        System.out.println("query code from " + url + "...");
        try {
            Thread.sleep((long) (Math.random() * 100));
        } catch (InterruptedException e) {
        }
        return "601857";
    }

    static Double fetchPrice(String code, String url) {
        System.out.println("query price from " + url + "...");
        try {
            Thread.sleep((long) (Math.random() * 100));
        } catch (InterruptedException e) {
        }
        return 5 + Math.random() * 20;
    }
}

anyOf()表示“任意个CompletableFuture只要一个成功”
allOf()表示“所有CompletableFuture都必须成功”

`ForkJoin`

一种基于“分治”的算法：通过分解任务，并行执行，最后合并结果得到最终结果。

写一个不严谨的例子

import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class Main {
    public static void main(String[] args) throws Exception {
        // 创建900000000个随机数组成的数组:
        long[] array = new long[100000000];
        long expectedSum = 0;
        for (int i = 0; i < array.length; i++) {
            array[i] = random();
            expectedSum += array[i];
        }
        System.out.println("Expected sum: " + expectedSum);
        // fork/join:
        ForkJoinTask<Long> task = new SumTask(array, 0, array.length);
        long startTime = System.currentTimeMillis();
        Long result = ForkJoinPool.commonPool().invoke(task);
        long endTime = System.currentTimeMillis();
        System.out.println("Fork/join sum: " + result + " in " + (endTime - startTime) + " ms.");
        long res = 0;
        startTime = System.currentTimeMillis();
        for (int i = 0; i < array.length; i++) {
            res += array[i];
        }
        endTime = System.currentTimeMillis();
        System.out.println("Common sum: " + res + " in " + (endTime - startTime) + " ms.");
    }

    static Random random = new Random(0);

    static long random() {
        return random.nextInt(10000);
    }
}

class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 500;
    long[] array;
    int start;
    int end;

    SumTask(long[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            // 如果任务足够小,直接计算:
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += this.array[i];
                // 故意放慢计算速度:
                //          try {
                //            Thread.sleep(1);
                //        } catch (InterruptedException e) {
                //        }
            }
            return sum;
        }
        // 任务太大,一分为二:
        int middle = (end + start) / 2;
        //System.out.println(String.format("split %d~%d ==> %d~%d, %d~%d", start, end, start, middle, middle, end));
        SumTask subtask1 = new SumTask(this.array, start, middle);
        SumTask subtask2 = new SumTask(this.array, middle, end);
        invokeAll(subtask1, subtask2);
        Long subresult1 = subtask1.join();
        Long subresult2 = subtask2.join();
        Long result = subresult1 + subresult2;
        //System.out.println("result = " + subresult1 + " + " + subresult2 + " ==> " + result);
        return result;
    }
}

`ThreadLocal`（线程本地变量）

其实就是为了避免冲突在每个线程中都为变量创建了一个副本

使用ThreadLocal要用try ... finally结构，并在finally中清除。

举例如下：

java

import java.util.stream.IntStream;

public class Main {
  public static void main(String[] args) {
    ThreadLocal<String> local = new ThreadLocal<>();
    IntStream.range(0, 10).forEach(i -> new Thread(() -> {
      try {
        local.set(Thread.currentThread().getName() + ":" + i);
        System.out.println("线程：" + Thread.currentThread().getName() + ",local:" + local.get());
      } finally {
        local.remove();
      }
    }).start());
  }
}

Java 多线程 ​

基本概念 ​

查看进程线程的方式 ​

Windows ​

Linux ​

Java ​

创建新线程 ​

一、从Thread派生一个自定义类，然后重写run()方法： ​

二、创建Thread实例时，传入一个Runnable实例： ​

三、使用lambda ​

线程状态 ​

Waiting状态 ​

Interrupted状态 ​

守护线程 ​

线程同步 ​

线程安全 ​

多线程常用辅助类 ​

CountDownLatch ​

CyclicBarrier ​

Semaphore ​

wait() 与 notify() ​

ReentrantLock ​

Condition ​

ReadWriteLock ​

StampedLock（乐观锁） ​

Concurrent ​

Atomic ​

ExecutorService（线程池） ​

Future ​

CompletableFuture ​

ForkJoin ​

ThreadLocal（线程本地变量） ​

Java 多线程

基本概念

查看进程线程的方式

Windows

Linux

Java

创建新线程

一、从`Thread`派生一个自定义类，然后重写`run()`方法：

二、创建`Thread`实例时，传入一个`Runnable`实例：

三、使用lambda

线程状态

Waiting状态

Interrupted状态

守护线程

线程同步

线程安全

多线程常用辅助类

CountDownLatch

CyclicBarrier

Semaphore

`wait()` 与 `notify()`

`ReentrantLock`

`Condition`

`ReadWriteLock`

`StampedLock`（乐观锁）

`Concurrent`

`Atomic`

`ExecutorService`（线程池）

Future

`CompletableFuture`

`ForkJoin`

`ThreadLocal`（线程本地变量）