多线程与并发原理

2020-02-18 本文已影响0人宣泽彬

为什么需要多线程

Java的执行模型是同步/阻塞（BLOCK）的。

默认情况下只有一个线程，因此需要多线程来提高效率。

如何简单创建一个线程

new Thread(new Runnable() {/** do something **/}).start()

需要注意的是，在执行start后，该线程便开始了它的工作。

但是当前线程会立即往下继续执行，而不会等待新的线程结束其任务。

线程注意事项

Java中只有Thread这么一种东西代表线程
只有执行了start方法之后才开始并发执行
每多开一个线程，就多一个执行流
方法栈是线程私有的（局部变量）
静态方法/类变量是被所有线程共享的
每一个新创建的线程最底下的方法栈为Thread.run()

多线程带来的麻烦的来源

切记：多线程的变量共享是所有坑的来源！

多线程难使用的原因是：你要看着同一份代码，想象着不同的人在疯狂地以乱序来执行它。

多线程使用的场合以及其带来的提升

适合多线程使用的场合：IO密集型应用极其有用 ---> 文件IO和网络IO（包括数据库）
不适合多线程使用的场合：CPU密集型（CPU intense），在这种情况下使用多线程带来的提升有限

多线程带来的性能提升的上限：

单核CPU ---> 100%
多核CPU ---> N * 100%

多线程不安全的表现

数据错误：经典例子 ---> i++
HashMap的死循环问题
死锁： synchronized的同步死锁问题
- 一般new Object()用于创建锁
- 排查死锁的方法和步骤:
  1. 利用命令：ps aux | grep java或java自带的命令jps来列出当前Java内的所有进程
  2. 使用jstack + 进程ID当前进程的所有信息
  3. 观察各个线程的调用栈，去发现死锁的问题
- 预防死锁产生的原则：所有的线程都按照相同的顺序获得资源的锁

实现线程安全的基本手段

使用不可变类：如Integer/String/...
使用synchronized同步块，但要搞清楚它同步了什么东西：
- synchronized + 一个对象，这种情况下是把这个对象当成了锁
- static synchronized + 方法，这种情况下是把这个class对象当成了锁，当这个类实例化之后，实例中的synchronized可变成synchronized (this) {...}，这里的this指当前这个实例化的对象
使用Collections下的synchronized处理之后的数据结构来替代ArrayList、HashSet、TreeSet、HashMap、LinkedHashMap这些线程不安全的数据结构
有时候使用上述的数据结构依然没能解决非原子操作的问题，这个时候需要使用JUC包（java.util.concurrent）下的方法：
- AtomicInteger/..: 可以把i++这些操作变成原子性的i.addAndGet(1)
- ConcurrentHashMap: 任何使用HashMap有线程安全问题的地方都可以无脑地使用ConcurrentHashMap进行代替
- ReentrantLock: 是一种可重入锁，与synchronized区别于它可以在一个地方加锁，在另一个地方进行解锁。

线程的历史

Java从一开始就把线程作为语言特性，提供了语言级别的智齿

因此为什么Java中的所有对象都可以成为锁：因为Object.wait()/notify()/notifyAll()方法提供了线程合作的可能性

Object.wait(): 释放锁，并阻塞线程等待唤醒
Object.notify(): 唤醒其中一个正在等待此监视器（锁）的线程，具体唤醒哪个由JVM决定
Object.notifyAll(): 唤醒所有等待此监视器（锁）的线程，并让它们自由竞争获得此监视器，没有获得的线程将继续等待

线程的六种状态

初始（NEW）
运行（RUNNABLE）
阻塞（BLOCKED）
等待（WAITING）
超时等待（TIME_WAITING）
终止（TERMINATED）

四种解决生产者/消费者模型的办法（具体用法请查对应文档）：

Object.wait/notify/notifyAll (Java生来就有)
Lock/Condition (JDK5 以后引入的)
BlockingQueue (阻塞队列)
其他第三方jar包引入的方法...

线程池

为什么需要线程池

因为线程是昂贵的，Java线程模型存在着缺陷，Java的线程调度完全依赖于操作系统的线程调度。

什么是线程池

线程池就是预先定义好的若干个线程

线程池类 Executors
固定数量线程池 newFixedThreadPool

线程池的submit方法和Thread的start方法十分类似，执行后不会等待，会立即向下执行，因此若需要获得新建线程返回的数据，需要使用Future类的get方法来阻塞获取。

最后分享一个word count的代码例子，可以在此代码中学习如何使用线程池：

package com.github.hcsp.multithread;

import java.io.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

public class MultiThreadWordCount1 {
    private static List<Future<HashMap<String, Integer>>> futures = new ArrayList<>();
    // 使用threadNum个线程，并发统计文件中各单词的数量
    public static Map<String, Integer> count(int threadNum, List<File> files) throws FileNotFoundException, ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(threadNum);
        for (File file : files) {
            BufferedReader reader = new BufferedReader(new FileReader(file));
            for (int i = 0; i < threadNum; i++) {
                Future<HashMap<String, Integer>> future = threadPool.submit(() -> getTheCountResult(file, reader));
                futures.add(future);
            }
        }

        return mergeTheFutureAndGetTheResult(futures);
    }

    private static HashMap<String, Integer> mergeTheFutureAndGetTheResult(List<Future<HashMap<String, Integer>>> futures) throws ExecutionException, InterruptedException {
        HashMap<String, Integer> finalResult = new HashMap<>();
        for (Future<HashMap<String, Integer>> future:futures) {
            for (Map.Entry<String, Integer> entry: future.get().entrySet()) {
                finalResult.put(entry.getKey(), finalResult.getOrDefault(entry.getKey(), 0) + entry.getValue());
            }
        }
        return finalResult;
    }

    private static HashMap<String, Integer> getTheCountResult(File file, BufferedReader reader) throws IOException {
        HashMap<String, Integer> result = new HashMap<>();
        String line;
        while ((line = reader.readLine()) != null) {
            String[] words = line.split(" ");
            for (String word : words) {
                result.put(word, result.getOrDefault(word, 0) + 1);
            }
        }
        return result;
    }
}