文章目錄

• 第十二章 并發(fā)
• • • `synchronized`關鍵字
    • 同步塊
    • 監(jiān)視器概念
    • `volatile`字段
    • `final`變量
    • 原子性
    • 死鎖
  • 線程安全的集合
  • 阻塞隊列
  • • 映射條目的原子更新
    • 對并發(fā)散列映射的批操作
    • 并行數(shù)組算法
    • 同步包裝器
  • 任務和線程池
  • • `Callable, Future`
    • `Executor`執(zhí)行器
    • `fork-in`框架
  • 異步計算
  • • 可完成`Future`
    • 組合可完成`Future`
  • 進程
  • • 建立一個進程

第十二章 并發(fā)

synchronized關鍵字

• 鎖用來保護代碼片段,一次只能有一個線程執(zhí)行被保護代碼
• 鎖可以管理試圖進入被保護代碼段的線程
• 一個鎖可以有一個或者多個相關聯(lián)的條件對象
• synchronized,設置內(nèi)部鎖
• 其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(...)} }

• 程序員用一個對象鎖實現(xiàn)額外的原子操作,稱為客戶端鎖定(十分脆弱)

---

監(jiān)視器概念

• 監(jiān)視器特性:

只包含私有字段

每個對象都有一個關聯(lián)的鎖

所有的方法由這個鎖鎖定 (客戶端調(diào)用object,method(), object對象的方法調(diào)用開始時自動獲得鎖,方法返回時自動釋放,這樣可以確保一個線程處理字段時,沒有其余的線程能夠訪問這些字段)

鎖可以有任意多個相關聯(lián)的條件

---

volatile字段

• 為實例字段的同步訪問提供了一中免鎖機制
• 下例使用內(nèi)部鎖.如果另一個線程已經(jīng)為該對象加鎖,兩種方法可能被阻塞

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分別以原子方式對一個整數(shù)進行自增或者自減

---

死鎖

• P584
• 當兩個線程互相操作且互相操作不能進行,形成死鎖

---

線程安全的集合

阻塞隊列

• 使用隊列可以安全的從一個線程向另一個線程傳遞數(shù)據(jù)
• 當試圖向隊列中添加元素且隊列已滿,或者從隊列中移除元素而隊列已空,阻塞隊列將導致線程阻塞
• put, take方法: 向隊列中添加或者移除元素,隊列滿或者空就阻塞,與不帶超時參數(shù)的方法offer, poll等效

---

映射條目的原子更新

• ConcurrentHashMap類中compute方法可以提供一個鍵和一個計算新值的函數(shù),比如可以更新一個整數(shù)計數(shù)器的映射

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

• computeIfAbsent, computeIfPresent分別只在沒有原值和有原值的情況下計算新值

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

• merge方法表示一個參數(shù)鍵不存在的時候提供初始值,存在就調(diào)用提供的函數(shù)結合初始值和原值

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

---

對并發(fā)散列映射的批操作

• 即使有其他線程在處理映射,批操作也能安全執(zhí)行
• 有search, reduce, forEach操作,可以operation, operationKeys, operationValues, operationEntries分別處理鍵和值,處理鍵,處理值,處理Map.Entry對象
• 需要指定一個參數(shù)化閾值,映射包含元素多于這個值就會進行批操作

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)

• 搜索超過一百次出現(xiàn)的單詞

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

并行數(shù)組算法

• P601

---

同步包裝器

• 任何集合類都可以通過使用同步包裝器變成線程安全的

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

---

任務和線程池

Callable, Future

• Runnable封裝一個異步運行的任務, 沒有參數(shù)和返回值的異步方法
• Callable有返回值,和前者類似,該接口是一個參數(shù)化類型,只有一個方法call

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

• Future保存異步計算的結果

• 執(zhí)行Callable方法之一是使用FutureTask. 它實現(xiàn)了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執(zhí)行器

• newCachedThreadPool構造一個線程池,必要時創(chuàng)建新線程,空閑狀態(tài)可以保持六十秒
• newFixedThreadPool構造一個具有固定大小的線程池;提交任務數(shù)量大于線程數(shù),放到隊列中,任務完成后再運行排隊的任務
• newSingleThreadExecutor只有一個線程的池,順序執(zhí)行提交的任務
• 并發(fā)線程數(shù)量=處理器內(nèi)核數(shù) 為最優(yōu)的運行速度
• 提交任務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對象會調(diào)用get來獲得值,這個方法會阻塞直到值可用
• CompletableFuture類實現(xiàn)了Future接口,注冊回調(diào),一旦結果可用,就會在某個線程中調(diào)用這個回調(diào),無需阻塞
• 被稱為可完成的,是因為可以手動設置一個完成值,這樣的對象在其他并發(fā)庫中稱為承諾

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

• 非阻塞調(diào)用通過回調(diào)來實現(xiàn)
• 程序員為任務完成后要出現(xiàn)的動作注冊一個回調(diào)
• 函數(shù)thenCompose組合函數(shù)T->CompletableFuture<U> 和U -> CompletableFuture<V>
• whenComplete方法用于處理異常,還有handle方法,他需要一個函數(shù)處理結果或者異常,并且計算一個新結果

---

進程

• Process類在一個單獨的操作系統(tǒng)中執(zhí)行一個命令,允許我們標準輸入.輸出和錯誤流交互
• ProcessBuilder類允許配置Process對象

---

建立一個進程

• 用命令和參數(shù)構建一個進程構建器

• var builder = new Process

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

• 第一個字符串必須是一個可執(zhí)行命令

• 改變工作目錄

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

• 處理進程的標準輸入流,輸出和錯誤流

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

• Attention: 進程的輸入流是jvm的一個輸出流,進程將輸出展示到控制臺上
• startPipeLine利用管道將一個進程的輸入作為另一個進程的輸出

---

總結

以上是生活随笔為你收集整理的CoreJava 笔记总结-第十二章 并发-2的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。