第12章并发

基本概念
线程的状态
- 线程状态转化图
InterruptedException 中断异常
- 代码
守护进程
uncaughtException 未捕获异常
- 代码
线程优先级
同步
线程安全的集合
高效并发操作
- ConcurrentHashMap的批操作
- 数组的并行操作
任务和线程池
异步计算
- CompletableFuture

基本概念

并发与并行

并发：轮流运行并行：多核同时执行

进程与线程

进程：程序执行一次的过程线程：进程中一个独立运行的路径

什么是多线程

多线程：一个进程同时完成多个任务，不同的任务由不同的线程独立执行

多线程与多进程

多进程：数据独立多线程：共享数据多线程通信更容易资源消耗更小

线程的状态

操作系统中有5总 java6种可以通过 t.getState() 获取

New(新建)

new Thread(r);

Runnable（可运行）

t.start();
此时等待CPU调度抢占时间片不一定马上被执行
Blocked（阻塞)、

尝试获取对象锁但该锁被占用

Waiting(等待)

等待其他线程通知
（如：调用wait join 等待Lock Condition）

Timed waiting(计时等待)

等待其他线程通知或超时
（如带有参数的 sleep wait join Lock.tryLock Condition.await）

Terminated(终止)

运行结束或抛异常或主动停止

线程状态转化图

InterruptedException 中断异常

一个等待状态的线程被调用interrupt()方法就会抛出所以 sleep wait 方法需要捕获中断异常

代码

public class Test {
    public static void main(String[] args) {
        //开启睡觉线程
        Thread t = new Thread(()->{
            try {
                Task.sleepTask();
            } catch (InterruptedException e) {
                //被打断时的处理
                System.out.println("睡觉线程被打断了 正在进行业务操作 如：数据回滚");
            }
        },"睡觉线程");
        t.start();
        //打断睡觉线程
        try {
            TimeUnit.SECONDS.sleep(2);
            System.out.println("主线程两秒后打断睡觉线程");
            t.interrupt();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
class Task{
    public static void sleepTask() throws InterruptedException {
        System.out.println(Thread.currentThread().getName()+" 开始睡觉了");
            TimeUnit.SECONDS.sleep(10);
    }
}

!注：InterruptedException 推荐抛出这样调用该的人才能处理

守护进程

t.setDaemon(true); 1.必须在调用前设置 2.一般用于给别的线程提供服务如：倒计时线程 GC线程 3.当只剩下守护线程时程序就结束

uncaughtException 未捕获异常

1.出现未捕获异常时线程死亡前会把异常传给UncaughtExceptionHandler的实现类处理 2.默认的处理器是ThreadGroup 如果没设置就是单纯的打印栈信息 3.给t1线程设置处理器 t1.setUncaughtExceptionHandler(UncaughtExceptionHandler eh); 4.给全部线程设置处理器 Thread.setDefaultUncaughtExceptionHandler(UncaughtExceptionHandler eh);

默认未捕获异常处理器：ThreadGroup的源码

代码

public static void main(String[] args) throws InterruptedException {
        Thread thread = new Thread(()->{
            int i = 1/0;
            System.out.println("OK");
        });
        thread.setUncaughtExceptionHandler((t,e)->{
            System.out.println(t.getName()+" 线程出现了未知异常："+e.getMessage()+"，正在记录日志");
        });
        thread.start();
    }

结果：

线程优先级

1 ~ 10 默认5 10最大 thread.setPriority(10); / thread.setPriority(Thread.MAX_PRIORITY);

同步

覆盖问题

多个线程同时修改一个对象发送了覆盖现象

覆盖问题例子

小红小明各有1000元存银行银行总存款为2000 让两人互相转钱结果出现银行总存款大于2000 各自余额小于0

代码

public class Bank {
    //银行账号
    private List<Account> accounts = new ArrayList<>();
    //开户
    public void openAccount(Account account) {
        accounts.add(account);
        System.out.println(account.getName() + "开户成功 存款：" + account.getBalance()
                + " 银行总存款：" + getTotalBalance());
    }
    //转账
    public void transfer(Account from, Account to, int amount) {
        from.setBalance(from.getBalance() - amount);
        to.setBalance(to.getBalance() + amount);
        System.out.println(from.getName() + " 给 " + to.getName() + " 转了：" + amount
                + " 当前余额：" + from.getBalance() + " 银行总存款：" + getTotalBalance());
    }
    //查询银行总存款
    public int getTotalBalance() {
        return accounts.stream().mapToInt(Account::getBalance).sum();
    }
    public static void main(String[] args) {
        Bank bank = new Bank();
        Account a1 = new Account("小明", 1000);
        Account a2 = new Account("小红", 1000);
        bank.openAccount(a1);
        bank.openAccount(a2);
        //开两个线程让小红小明互相转钱
        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                while (true) {
                    bank.transfer(a1, a2, 500);
                    bank.transfer(a2, a1, 500);
                }
            }).start();
        }
    }
}
class Account {
    private String name;
    private int balance;
    public Account(String name, int balance) {
        this.name = name;
        this.balance = balance;
    }
    public String getName() {
        return name;
    }
    public int getBalance() {
        return balance;
    }
    public void setBalance(int balance) {
        this.balance = balance;
    }
}

原因及图解

转账的操作不具有原子性例如：线程1 和线程2 同时执行给账号A转账100的操作由于转账操作不具有原子性线程2在线程1将本地内存中的数据写回共享内存前先完成了转账操作结果线程2的操作结果被线程1给覆盖了理论存款为1200 实际只有1100

解法1 重入锁ReentrantLock

修改转账和查询方法在方法前后加上重入锁将方法变为原子性操作

在转账前加入余额不足等待操作在转账后唤醒等待的账号

重入锁：

线程可以嵌套获得锁对象比如AB方法都有锁 A调用B 就是锁了两次
请确保调用的是同一个Lock对象锁才生效
解锁推荐放在finally里
有加锁就必须解锁否则会发送死锁
可以配置公平锁（先来后到）

Condition

判断条件要放在while里不然会产生虚假唤醒
唤醒最好用signalAll方法 signal只唤醒一个如果唤醒的这个不符合条件就会产生死锁

public class Bank {
    //银行账号
    private List<Account> accounts = new ArrayList<>();
    //重入锁
    private Lock lock = new ReentrantLock();
    //余额不足等待条件
    private Condition sufficientFunds = lock.newCondition();
    //开户
    public void openAccount(Account account) {
        accounts.add(account);
        System.out.println(account.getName() + "开户成功 存款：" + account.getBalance()
                + " 银行总存款：" + getTotalBalance());
    }
    //转账
    public void transfer(Account from, Account to, int amount) {
        lock.lock();
        try {
            while (from.getBalance() < amount)
                sufficientFunds.await();
            from.setBalance(from.getBalance() - amount);
            to.setBalance(to.getBalance() + amount);
            sufficientFunds.signalAll();
            System.out.println(from.getName() + " 给 " + to.getName() + " 转了：" + amount
                    + " 当前余额：" + from.getBalance() + " 银行总存款：" + getTotalBalance());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    //查询银行总存款
    public int getTotalBalance() {
        lock.lock();
        try {
            return accounts.stream().mapToInt(Account::getBalance).sum();
        } finally {
            lock.unlock();
        }
    }
    public static void main(String[] args) {
        Bank bank = new Bank();
        Account a1 = new Account("小明", 1000);
        Account a2 = new Account("小红", 1000);
        bank.openAccount(a1);
        bank.openAccount(a2);
        //开两个线程让小红小明互相转钱
        for (int i = 0; i < 4; i++) {
            new Thread(() -> {
                while (true) {
                    bank.transfer(a1, a2, 500);
                    bank.transfer(a2, a1, 500);
                }
            }).start();
        }
    }
}

解法2 synchronized 同步方法

调用synchronized锁住转账和查询方法

加入余额不足等待和唤醒

synchronized

锁方法就是谁调用锁谁如果是静态方法就是锁类对象的锁 A.class

与Lock的区别

不能中断
不能指定超时时间（lock有tryLock方法）
只有一个条件 this.wait 可能不够用

public class Bank {
    //银行账号
    private List<Account> accounts = new ArrayList<>();
    //开户
    public void openAccount(Account account) {
        accounts.add(account);
        System.out.println(account.getName() + "开户成功 存款：" + account.getBalance()
                + " 银行总存款：" + getTotalBalance());
    }
    //转账
    public synchronized void transfer(Account from, Account to, int amount){
        while (from.getBalance() < amount) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        from.setBalance(from.getBalance() - amount);
        to.setBalance(to.getBalance() + amount);
        this.notifyAll();
        System.out.println(from.getName() + " 给 " + to.getName() + " 转了：" + amount
                + " 当前余额：" + from.getBalance() + " 银行总存款：" + getTotalBalance());
    }
    //查询银行总存款
    public synchronized int getTotalBalance() {
        return accounts.stream().mapToInt(Account::getBalance).sum();
    }
    public static void main(String[] args) {
        Bank bank = new Bank();
        Account a1 = new Account("小明", 1000);
        Account a2 = new Account("小红", 1000);
        bank.openAccount(a1);
        bank.openAccount(a2);
        //开两个线程让小红小明互相转钱
        for (int i = 0; i < 4; i++) {
            new Thread(() -> {
                while (true) {
                    bank.transfer(a1, a2, 500);
                    bank.transfer(a2, a1, 500);
                }
            }).start();
        }
    }
}

解法3 同步代码块

public class Bank {
    //银行账号
    private List<Account> accounts = new ArrayList<>();
    //监视器
    private Object monitor = new Object();
    //开户
    public void openAccount(Account account) {
        accounts.add(account);
        System.out.println(account.getName() + "开户成功 存款：" + account.getBalance()
                + " 银行总存款：" + getTotalBalance());
    }
    //转账
    public void transfer(Account from, Account to, int amount) {
        synchronized (monitor) {
            while (from.getBalance() < amount) {
                try {
                    monitor.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            from.setBalance(from.getBalance() - amount);
            to.setBalance(to.getBalance() + amount);
            monitor.notifyAll();
            System.out.println(from.getName() + " 给 " + to.getName() + " 转了：" + amount
                    + " 当前余额：" + from.getBalance() + " 银行总存款：" + getTotalBalance());
        }
    }
    //查询银行总存款
    public int getTotalBalance() {
        synchronized (monitor) {
            return accounts.stream().mapToInt(Account::getBalance).sum();
        }
    }
    public static void main(String[] args) {
        Bank bank = new Bank();
        Account a1 = new Account("小明", 1000);
        Account a2 = new Account("小红", 1000);
        bank.openAccount(a1);
        bank.openAccount(a2);
        //开两个线程让小红小明互相转钱
        for (int i = 0; i < 4; i++) {
            new Thread(() -> {
                while (true) {
                    bank.transfer(a1, a2, 500);
                    bank.transfer(a2, a1, 500);
                }
            }).start();
        }
    }
}

volatile

保证线程可见性

如下代码不加volatile线程将不会停止

private volatile static boolean flag = true;
    public static void main(String[] args) throws InterruptedException {
        new Thread(()->{
            while (flag){
            }
            System.out.println("循环停止");
        }).start();
        TimeUnit.SECONDS.sleep(1);
        flag = false;
    }

2.防止指令重排 DCL双重检测锁必须加volatile 或者说多线程下的判断条件最好都加

原子操作

//AtomicInteger
int result = atomicInteger.incrementAndGet();//+1
atomicInteger.compareAndSet(result, 0);//设为0
atomicInteger.updateAndGet((x)->{return Math.max(x, 10);});//和10比较
atomicInteger.accumulateAndGet(3,(x,y)->{return x+y;});//+3
//LongAdder 提升效率的自增 分开计算最后合并
LongAdder adder = new LongAdder();
   for (int i = 0; i < 1000; i++) {
       new Thread(()->{
           adder.increment();
       }).start();
   }
System.out.println(adder.sum());
//LongAccumulator 表达式操作
LongAccumulator accumulator = new LongAccumulator(Math::max,0);
   for (int i = 0; i < 1000; i++) {
      new Thread(()->{
         accumulator.accumulate(1);
      }).start();
    }
System.out.println(accumulator.get());

线程局部变量 ThreadLocal

可以保证线程持有一个本地对象原理是一个Map key为线程本身 value是持有的对象 key和value通过弱引用连接的一起方便线程消销毁时回收对象

应用： SimpleDateFormat线程不安全通过ThreadLocal保证其安全

Random线程安全但是多线程会降低效率 ThreadLocalRandom.current().nextInt(10)；可提高效率

public static ThreadLocal<SimpleDateFormat> string = ThreadLocal
            .withInitial(() -> {
                return new SimpleDateFormat("yyy-MM-dd");
            });

死锁

多个线程互相等待对方锁持有的锁而导致程序停住 java语言本身不能解决死锁必须在程序设计层面解决

死锁是狄杰斯克拉提出来了有四个必要条件：条件互斥请求与保持不可剥夺循环等待破环其中至少一种条件就可解决死锁

常用解法：管程法信号灯法

为什么弃用 stop suspend方法

这两个方法容易造成死锁

线程安全的集合

阻塞队列

阻塞队列：队满时插入和队空时获取时可阻塞的队列

当发送队满时插入和队空时获取时的三种方法：

阻塞

put / take 适合做线程管理

返回值

offer / peek / poll offer和poll可以指定时间

抛异常

add / element / remove

ArrayBlockingQueue 可以指定是否公平 LinkedBlockingQueue 理论上大小无上线并且是一个双端队列 PriorityBlockingQueue 可以指定优先级 DelayQueue 延迟队列时间到了才能获取比如完成3秒后执行提交的动作 TransferQueue 传递队列生产者调用传递方法等待消费者接收可以完成聊天功能

！注：当poll和peek失败时返回null 所以阻塞队列不能存放null值

//DelayQueue事例
public class Test {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue queue = new DelayQueue();
        queue.put(new MyDelayed(() -> {
            System.out.println("执行任务1");
        }, 3, TimeUnit.SECONDS));
        queue.put(new MyDelayed(() -> {
            System.out.println("执行任务2");
        }, 6, TimeUnit.SECONDS));
        queue.put(new MyDelayed(() -> {
            System.out.println("执行任务3");
        }, 3, TimeUnit.SECONDS));
        while (true) {
            MyDelayed task = (MyDelayed) queue.take();
            task.execute();
        }
    }
}
//自定义延迟对象
class MyDelayed implements Delayed {
private long executionTime;
    //时间到了后执行的任务
    private Runnable runnable;
    public MyDelayed(Runnable runnable, long delay, TimeUnit timeUnit) {
        this.runnable = runnable;
        this.executionTime = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(delay, timeUnit);
    }
    //获得过去时间 大于 0 则过期
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert( executionTime - System.currentTimeMillis() , TimeUnit.MILLISECONDS);
    }
    //队列内部比较 优先返回快过期的任务
    @Override
    public int compareTo(Delayed o) {
        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    }
    public void execute() {
        new Thread(runnable).start();
    }
}

//TransferQueue 生产者等待消费者上线
public static void main(String[] args) throws InterruptedException {
        TransferQueue queue = new LinkedTransferQueue();
        for (int i = 0; i < 10; i++) {
            int tmp = i;
            new Thread(() -> {
                try {
                    queue.transfer("消费者传过来的数据" + tmp);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
        new Thread(() -> {
            while (true) {
                try {
                    String msg = (String) queue.take();
                    System.out.println("我是消费者 我接收到了：" + msg);
                    TimeUnit.MILLISECONDS.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

阻塞队列例子

判断文件夹下的文件是否包含某个关键字

一个线程读取文件夹下面的全部文件放入阻塞队列中最后一个文件用特殊标志位代替另一个线程监听接收队列中的文件并开启新的线程进行查找操作当监听到特殊标志位文件时结束

/**
 * 类描述： 利用阻塞队列和多线程 读取文件夹下代码的行数
 */
public class CountCodeLines {
    //文件夹路径
    private BlockingQueue<File> queue = new LinkedBlockingQueue<>();
    private static Long lines = 0L;
    private static AtomicInteger finished = new AtomicInteger();
    private File dummy = new File("");//用于做结束的判断条件
    //线程池
    private ExecutorService threadPool = new ThreadPoolExecutor(2,
            Runtime.getRuntime().availableProcessors(),
            30,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(30),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.CallerRunsPolicy());
    /**
     * 统计代码量
     *
     * @return
     */
    public long count(String dir) throws InterruptedException {
        //获取所有文件放入阻塞队列
        threadPool.execute(()->{doReadFiles(dir);});
        //监听阻塞队列并统计
        int count = 0;
        do {
            File file = queue.take();
            System.out.println(file.getAbsolutePath());
            if (file.equals(dummy)) {
                break;
            }
            //开启线程读取行数
            threadPool.execute(()->{readLines(file);});
            count++;
        } while (true);
        threadPool.shutdown();
        while (count != finished.get()) {
            TimeUnit.MILLISECONDS.sleep(100);
        }
        return lines;
    }
    //将java文件从文件夹中读出来并放入阻塞队列
    private void doReadFiles(String dir) {
        try {
            readFiles(dir);
            queue.put(dummy);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    //将java文件从文件夹中读出来并放入阻塞队列
    private void readFiles(String dir) throws InterruptedException {
        File file = new File(dir);
        if (file.isDirectory()) {
            String[] files = file.list();
            for (String subFile : files) {
                //递归
                this.readFiles(file.getAbsolutePath() + "\\" + subFile);
            }
        } else if ("java".equals(file.getName().substring(file.getName().lastIndexOf(".") + 1))) {
            queue.put(file);
        }
    }
    //读取每个文件的行数 并加到共享变量中
    private void readLines(File file) {
        try {
            int tmp = 0;
            Scanner scanner = new Scanner(file);
            while (scanner.hasNextLine()) {
                scanner.nextLine();
                tmp++;
            }
            synchronized (lines) {
                lines += tmp;
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }finally {
            scanner.close();
        }
        finished.incrementAndGet();
    }
    public static void main(String[] args) {
        CountCodeLines countCodeLines = new CountCodeLines();
        try {
            System.out.println(countCodeLines.count("D:\\eclipse-workspace"));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

并发的Map Set Queue

ConcurrentHashMap ConcurrentSkipListMap ConcurrentSkipListSet ConcurrentLinkedQueue

这些集合采用了复杂的算法尽可能提高并发访问时的效率他们的size方法是通过变量合并的不是顺发的 mappinngCount方法可以返回Long类型的个数防止21亿不够用返回的迭代器是弱一致性的就是不一定能反应之后的修改情况但是不会抛出并发修改异常

ConcurrentHashMap 默认支持16个并发大于16就阻塞不允许null值 ConcurrentLinkedQueue 线程安全无上限 ConcurrentSkipListSet 跳表数据结构有序元素需要实现Comparable ConcurrentSkipListMap 键有序同上

ConcurrentHashMap 的键可以当并发的set Set set = ConcurrentHashMap.newKeySet();

//保证原子性的自增
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap();
map.compute("key",(k,v)->{return v == null ? 1 : v + 1;});
System.out.println(map.get("key"));//结果为1
map.merge("key", 2, (x,y)->{return x+y;});
System.out.println(map.get("key"));//结果为3

CopyOnWrite

修改时先拷贝一个新的副本用新的副本修改后替换掉老的这样查询的时候就不用上锁了当查询远大于修改时很推荐使用但有个缺点：修改时正在遍历的线程可能会访问到老的数据

CopyOnWriteArrayList list = new CopyOnWriteArrayList(); CopyOnWriteArraySet set = new CopyOnWriteArraySet();

Collections提供的线程方法

List<Object> list = Collections.synchronizedList(new ArrayList<>());
Map<Object, Object> map = Collections.synchronizedMap(new HashMap<>());
Set<Object> set = Collections.synchronizedSet(new HashSet<>());

Collections 给每个子类都提供了支持并发的内部包装类本质上就是返回包装类对象该对象大部分方法都上锁如map：

当经常修改时用Collections的方法比较好查询比修改多很多时用CopyOnWrite比较好但是Collections的内部类依然不能遍历时修改这时要么自己上锁要么用ConcurrentHashMap或者CopyOnWrite

总结

需要阻塞就用阻塞队列
遍历时修改就用Concurrent的和 CopyOnWrite 不然会抛并发修改异常

但是这两个都不能保证遍历的数据是最新的

Collections提供的内部类能保证单一的增删改查是原子性的
如果查操作多于增删改就用 CopyOnWrite 因为这玩意查询方法不用上锁比较快

高效并发操作

ConcurrentHashMap的批操作

有三种方法 search reduce forEach 搜索归约遍历

//search方法
//需要指定阀值 就是每个线程大约处理多少数据
//方法判断失败时返回null
//操作键 返回第一个长度大于10的键
String key = map.searchKeys(1000, (k)->{return k.length() > 10 ? k : null;});
//操作值 返回第一个大于1000的值
Integer value = map.searchValues(1000, (v)->{return v > 1000 ? v : null;});
//操作键和值 返回第一个值大于1000的键
String key = map.search(1000, (k, v) -> {return v > 1000 ? k : null;});
//操作键值对 同上
String key = map.searchEntries(1000, (e)->{return e.getValue() > 1000 ? e.getKey() : null;});
//reduce方法
//操作键和值 将值大于1000的数累加
Integer reduce = map.reduce(1000,
                (k, v) -> {return v > 1000 ? v : null;},
                (v1, v2) -> {return v1 + v2;});
//多了个默认值 当集合为空的时候返回0 但是遍历无结果仍然返回null
Integer reduce = map.reduceToInt(1000,
                (k, v) -> {return v > 100000 ? v : null;},
                 0,
                (v1, v2) -> {return v1 + v2;});

        //普通遍历
        map.forEach((k, v) -> {
            System.out.println(k + " " + v);
        });
        //并行
        map.forEach(1000, (k, v) -> {
            System.out.println(k + " " + v);
        });
        //第二个函数的返回值传给第三个函数作为参数
        map.forEach(1000, (k, v) -> {
            return k + " " + v;
        }, (s) -> {
            System.out.println(s);
        });

数组的并行操作

并行排序

Arrays.parallelSort(arr); //并行排序本质是ForkJoin

并行填充

Arrays.parallelSetAll(arr, (i)->{return i+1;}); 往arr中填写1 2 3 4 … 等数据

并行与前一个操作

Arrays.parallelPrefix(ints, (x, y) -> {return x + y;}); 与前一个相加比如 [ 2,5,1,0] 执行后 -> [2,7,8,8]

//填充ints 数组 1 2 3 4 5 6 7 8 9 10 
int[] ints = new int[10];
Arrays.parallelSetAll(ints, (i)->{return i+1;});
//与前一个相加
//比如 [ 2,5,1,0] 执行后 -> [2,7,8,8]
Arrays.parallelPrefix(ints, (x, y) -> {return x + y;});

并行填充本质：

任务和线程池

线程池：利益池化技术管理线程强调一个复用的概念能达到节约资源提高效率控制最大并发量的效果阿里巴巴开发手册不让使用Java自带的线程池对象因为默认最大值是Int的最大值可能会发生OOM 因此我们深刻理解线程池参数的含义掌握自己配置的能力

Callable和Future

callable 是有返回值的任务但是线程只能接收Runnable对象于是用Future封装了下并且通过Future管理（获取结果、取消任务、查询任务状态）

Callable<String> task = ()->{return "我是Callable的call方法";};
FutureTask<String> futureTask = new FutureTask<>(task);
new Thread(futureTask).start();
System.out.println(futureTask.get());

Executors

Executors可以通过静态工厂的方法创建线程池线程池可以执行Callable或Runnable任务并返回Future对象如果是调用Runnable任务返回的Future 调用get时只会简单的返回null 不用时请调用shoutdown或shoutdownNow

newCachedThreadPool
- 任务立即执行有空闲线程就调用没有就创建线程或过期
- 适合任务快或者任务阻塞
newFixedThreadPool
- 固定线程线程不够了就把任务放入等待队列
newSingleThreadExecutor
- 同上可以做性能分析通单线程替换然后统计运行时间
newScheduledThreadPool
- 固定线程延迟执行或延迟后周期执行
newSingleThreadScheduledExecutor
- 单线程同上

//定时
ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(5);
threadPool.schedule(() -> {
      System.out.println("延迟执行");
}, 5, TimeUnit.SECONDS);
threadPool.scheduleAtFixedRate(() -> {
      System.out.println("延迟然后周期执行执行");
},2, 3, TimeUnit.SECONDS);
threadPool.scheduleWithFixedDelay(() -> {
      System.out.println("延迟然后周期执行 上一个任务完成时才能调度");
}, 3,5, TimeUnit.SECONDS);

invokeAny invokeAll 和 ExecutorCompletionService

invokeAny：并发执行任务返回第一个结果并且返回后其他任务便不往下执行了适合完成搜索功能 invokeAll：并发执行任务等全部执行完后一切返回结果 ExecutorCompletionService：并发执行任务通过包装普通执行器创建会将执行的结果按序放到阻塞队列中一个一个获取

//返回第一个
String firstResult = threadPool.invokeAny(tasks);
System.out.println(firstResult);
//等全部完成 返回全部
List<Future<String>> futures = threadPool.invokeAll(tasks);
for (Future<String> future : futures) {
   System.out.println(future.get());
}
//完成一个返回一个
ExecutorCompletionService<String> service = new ExecutorCompletionService<>(threadPool);
for (Callable<String> task : tasks) {
    service.submit(task);
}
for (int i = 0; i < tasks.size(); i++) {
    System.out.println(service.take().get());
}

fork-join

类似递归的思想把大任务分成小任务多线程处理最后合并适合大数据计算有一个任务窃取的概念任务队列是双向的执行完任务的空闲线程会从末尾偷取别的线程的任务来工作使用方法继承RecursiveTask 或者 RecursiveAction 前者有返回值

//处理double数组 但测试感觉不是很快
class Counter extends RecursiveTask<Integer> {
    public static final int THRESHOLD = 10000;
    private double[] values;
    private int from;
    private int to;
    private DoublePredicate filter;
    public Counter(double[] values, int from, int to, DoublePredicate filter) {
        this.values = values;
        this.from = from;
        this.to = to;
        this.filter = filter;
    }
    @Override
    protected Integer compute() {
        if ((to - from) < THRESHOLD) {
            int count = 0;
            for (int i = from; i < to; i++) {
                if (filter.test(values[i])) count++;
            }
            return count;
        } else {
            //fork-join
            int mid = (from + to) / 2;
            Counter first = new Counter(values, from, mid, filter);
            Counter second = new Counter(values, mid, to, filter);
            invokeAll(first, second);
            return first.join() + second.join();
        }
    }
}

invockAll方法是父类的方法执行多个任务调用后阻塞直到所有任务都完成

异步计算

不用再阻塞等待结果通过回调方法异步处理

CompletableFuture

通过回调函数异步执行并且可以编写执行顺序

ExecutorService threadPool = Executors.newFixedThreadPool(4);
        // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 1.0;},threadPool);
        // 如果执行成功:
        cf.thenAccept((result) -> {
            System.out.println("price: " + result);
        });
        // 如果执行异常:
        cf.exceptionally((e) -> {
            e.printStackTrace();
            return null;
        });
        threadPool.shutdown();

第12章 并发

基本概念

并发与并行

进程与线程

什么是多线程

多线程与多进程

线程的状态

线程状态转化图

InterruptedException 中断异常

代码

守护进程

uncaughtException 未捕获异常

代码

线程优先级

同步

覆盖问题

覆盖问题例子

代码

原因及图解

解法1 重入锁ReentrantLock

解法2 synchronized 同步方法

解法3 同步代码块

volatile

原子操作

线程局部变量 ThreadLocal

死锁

为什么弃用 stop suspend方法

线程安全的集合

阻塞队列

阻塞队列例子

并发的Map Set Queue

CopyOnWrite

Collections提供的线程方法

总结

高效并发操作

ConcurrentHashMap的批操作

数组的并行操作

任务和线程池

Callable和Future

Executors

invokeAny invokeAll 和 ExecutorCompletionService

fork-join

异步计算

CompletableFuture