第十二章 并发

synchronized关键字

• 锁用来保护代码片段,一次只能有一个线程执行被保护代码
• 锁可以管理试图进入被保护代码段的线程
• 一个锁可以有一个或者多个相关联的条件对象
• synchronized,设置内部锁
• 其wait方法将线程增加到等待集中
• notifyAll, notify方法可以接触等待线程的阻塞

---

同步块

• 会获得obj锁

synchronized(obj){critical section
}

• public class Bank{private double[] accounts;private var lock = new Object();public void transfer(int from, int to, int amount){synchronized(lock){accounts[from] -= amount;accounts[to] += amount;}System.out.println(...)}
  }
  

• 程序员用一个对象锁实现额外的原子操作,称为客户端锁定(十分脆弱)

---

监视器概念

• 监视器特性:

1. 只包含私有字段
2. 每个对象都有一个关联的锁
3. 所有的方法由这个锁锁定 (客户端调用object,method(), object对象的方法调用开始时自动获得锁,方法返回时自动释放,这样可以确保一个线程处理字段时,没有其余的线程能够访问这些字段)
4. 锁可以有任意多个相关联的条件

---

volatile字段

• 为实例字段的同步访问提供了一中免锁机制
• 下例使用内部锁.如果另一个线程已经为该对象加锁,两种方法可能被阻塞

private boolean done;
public synchronized boolean isDone(){ return done;}
public synchronized void setDone() {done = true;}

---->

private volatile boolean done;
public boolean isDone(){ return done;}
public void setDone() {done = true;}

---

final变量

• final var accounts = new HashMap<String, Double>();其他线程会在构造器完成之后才看到accounts变量

---

原子性

• 对共享变量除了赋值不做其他操作,可以声明为volatile
• java.util.concurrent.atomic包中很多类保证原子性操作
• AtomicInteger类提供方法incrementAndGet, decrementAndGet分别以原子方式对一个整数进行自增或者自减

---

死锁

• P584
• 当两个线程互相操作且互相操作不能进行,形成死锁

---

线程安全的集合

阻塞队列

• 使用队列可以安全的从一个线程向另一个线程传递数据
• 当试图向队列中添加元素且队列已满,或者从队列中移除元素而队列已空,阻塞队列将导致线程阻塞
• put, take方法: 向队列中添加或者移除元素,队列满或者空就阻塞,与不带超时参数的方法offer, poll等效

---

映射条目的原子更新

• ConcurrentHashMap类中compute方法可以提供一个键和一个计算新值的函数,比如可以更新一个整数计数器的映射

map.compute(word, (k, v) -> v == null ? 1 : v + 1);

• computeIfAbsent, computeIfPresent分别只在没有原值和有原值的情况下计算新值

map.compyteIfAbsent(word, k -> new LongAdder()).increment();

• merge方法表示一个参数键不存在的时候提供初始值,存在就调用提供的函数结合初始值和原值

map.merge(word, 1L, (existingValue, newValue) -> existingValue+newValue);
//简便形式
map.merge(word, 1L, Long::sum);

---

对并发散列映射的批操作

• 即使有其他线程在处理映射,批操作也能安全执行
• 有search, reduce, forEach操作,可以operation, operationKeys, operationValues, operationEntries分别处理键和值,处理键,处理值,处理Map.Entry对象
• 需要指定一个参数化阈值,映射包含元素多于这个值就会进行批操作

U searchKeys(long threshold, BiFunction<? super K, ? extends U> f)
U searchValues(long threshold, BiFunction<? super K, ? extends U> f)
U search(long threshold, BiFunction<? super K, ? extends U> f)
U searchEntries(long threshold, BiFunction<? super K, ? extends U> f)

• 搜索超过一百次出现的单词

String result = map.search(threshold, (k, v) -> v > 1000 ? k : null);

并行数组算法

• P601

---

同步包装器

• 任何集合类都可以通过使用同步包装器变成线程安全的

List<E> synchArray  = Collections.synchronizedList(new ArrayList<E>());
Map<K, V> synchHashMap = Collections.synchroniedHashMap(new HashMap<K, V>());

---

任务和线程池

Callable, Future

• Runnable封装一个异步运行的任务, 没有参数和返回值的异步方法
• Callable有返回值,和前者类似,该接口是一个参数化类型,只有一个方法call

public interface Callable<V>{V call() throws Exception;
}

• Future保存异步计算的结果

• 执行Callable方法之一是使用FutureTask. 它实现了Future, Runnable接口,所以可以构造一个线程运行该任务

Callable<Integer> task = ;
var FutureTask = new FutureTask<Integer>(task);
var t = new Thread(futureTask);
t.start();
...
Integer result = task.get(); //it's a Future

---

Executor执行器

• newCachedThreadPool构造一个线程池,必要时创建新线程,空闲状态可以保持六十秒
• newFixedThreadPool构造一个具有固定大小的线程池;提交任务数量大于线程数,放到队列中,任务完成后再运行排队的任务
• newSingleThreadExecutor只有一个线程的池,顺序执行提交的任务
• 并发线程数量=处理器内核数 为最优的运行速度
• 提交任务Future<T> submit(Callable<T>, task)
• 结束线程池shutdown

    public static void main(String[] args)throws InterruptedException, ExecutionException, IOException{try (var in = new Scanner(System.in)){System.out.print("Enter base directory (e.g. /opt/jdk-9-src): ");String start = in.nextLine();System.out.print("Enter keyword (e.g. volatile): ");String word = in.nextLine();Set<Path> files = descendants(Path.of(start));var tasks = new ArrayList<Callable<Long>>();for (Path file : files){Callable<Long> task = () -> occurrences(word, file);tasks.add(task);}ExecutorService executor = Executors.newCachedThreadPool();// use a single thread executor instead to see if multiple threads// speed up the search// ExecutorService executor = Executors.newSingleThreadExecutor();Instant startTime = Instant.now();List<Future<Long>> results = executor.invokeAll(tasks);long total = 0;for (Future<Long> result : results)total += result.get();Instant endTime = Instant.now();System.out.println("Occurrences of " + word + ": " + total);System.out.println("Time elapsed: "+ Duration.between(startTime, endTime).toMillis() + " ms");var searchTasks = new ArrayList<Callable<Path>>();for (Path file : files)searchTasks.add(searchForTask(word, file));Path found = executor.invokeAny(searchTasks);System.out.println(word + " occurs in: " + found);if (executor instanceof ThreadPoolExecutor) // the single thread executor isn'tSystem.out.println("Largest pool size: "+ ((ThreadPoolExecutor) executor).getLargestPoolSize());executor.shutdown();}}

---

fork-in框架

• P612

---

异步计算

可完成Future

• 当有一个Future对象会调用get来获得值,这个方法会阻塞直到值可用
• CompletableFuture类实现了Future接口,注册回调,一旦结果可用,就会在某个线程中调用这个回调,无需阻塞
• 被称为可完成的,是因为可以手动设置一个完成值,这样的对象在其他并发库中称为承诺

var f = new CompletableFuture<Integer>();
excutor.excute(() -> {int n = workHard(arg);f.complete(n);
});
excutor.excute(() -> {int n = workSmart(arg);f.complete(n);
}) ;

• 对一个异常完成Future

Throwable t = ...;
f.completeExceptionally(t);

---

组合可完成Future

• 非阻塞调用通过回调来实现
• 程序员为任务完成后要出现的动作注册一个回调
• 函数thenCompose组合函数T->CompletableFuture<U> 和U -> CompletableFuture<V>
• whenComplete方法用于处理异常,还有handle方法,他需要一个函数处理结果或者异常,并且计算一个新结果

---

进程

• Process类在一个单独的操作系统中执行一个命令,允许我们标准输入.输出和错误流交互
• ProcessBuilder类允许配置Process对象

---

建立一个进程

• 用命令和参数构建一个进程构建器

• var builder = new Process

• Builder("gcc", "myapp.c");

• 第一个字符串必须是一个可执行命令

• 改变工作目录

builder = builder.directory(path.toFile());

• 处理进程的标准输入流,输出和错误流

OutputStream processIn = p.getOutputStream();
IntputStream processOut = p.getInputStream();
IntputStream processErr = p.getErrorStream();

• Attention: 进程的输入流是jvm的一个输出流,进程将输出展示到控制台上
• startPipeLine利用管道将一个进程的输入作为另一个进程的输出

---