并發及并發的執行緒安全處理
- 高并發處理的思路及手段
1 基本概念
1.1 并發
同時擁有兩個或者多個執行緒，如果程式在單核處理器上運行多個執行緒將交替地換入或者換出記憶體,這些執行緒是同時“存在"的，每個執行緒都處于執行程序中的某個狀態，如果運行在多核處理器上,此時，程式中的每個執行緒都將分配到一個處理器核上，因此可以同時運行.
##1.2 高并發( High Concurrency)
互聯網分布式系統架構設計中必須考慮的因素之一，通常是指，通過設計保證系統能夠同時并行處理很多請求.
##1.3 區別與聯系
- 并發: 多個執行緒操作相同的資源，保證執行緒安全，合理使用資源
- 高并發:服務能同時處理很多請求，提高程式性能
2 CPU
2.1 CPU 多級快取
- 為什么需要CPU cache
CPU的頻率太快了，快到主存跟不上
如此,在處理器時鐘周期內，CPU常常需要等待主存，浪費資源，所以cache的出現，是為了緩解CPU和記憶體之間速度的不匹配問題(結構:cpu-> cache-> memory ). - CPU cache的意義
- 時間區域性
如果某個資料被訪問，那么在不久的將來它很可能被再次訪問 - 空間區域性
如果某個資料被訪問，那么與它相鄰的資料很快也可能被訪問
- 時間區域性
2.2 快取一致性(MESI)
用于保證多個 CPU cache 之間快取共享資料的一致
- M-modified被修改
該快取行只被快取在該 CPU 的快取中,并且是被修改過的,與主存中資料是不一致的,需在未來某個時間點寫回主存,該時間是允許在其他CPU 讀取主存中相應的記憶體之前,當這里的值被寫入主存之后,該快取行狀態變為 E - E-exclusive獨享
快取行只被快取在該 CPU 的快取中,未被修改過,與主存中資料一致
可在任何時刻當被其他 CPU讀取該記憶體時變成 S 態,被修改時變為 M態 - S-shared共享
該快取行可被多個 CPU 快取,與主存中資料一致 - I-invalid無效
- 亂序執行優化
處理器為提高運算速度而做出違背代碼原有順序的優化
并發的優勢與風險
3 專案準備
3.1 專案初始化
3.2 并發模擬-Jmeter壓測
3.3 并發模擬-代碼
CountDownLatch
###Semaphore(信號量)
以上二者通常和執行緒池搭配
下面開始做并發模擬
package com.mmall.concurrency;
import com.mmall.concurrency.annoations.NotThreadSafe;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
/**
* @author JavaEdge
* @date 18/4/1
*/
@Slf4j
@NotThreadSafe
public class ConcurrencyTest {
/**
* 請求總數
*/
public static int clientTotal = 5000;
/**
* 同時并發執行的執行緒數
*/
public static int threadTotal = 200;
public static int count = 0;
public static void main(String[] args) throws Exception {
//定義執行緒池
ExecutorService executorService = Executors.newCachedThreadPool();
//定義信號量,給出允許并發的執行緒數目
final Semaphore semaphore = new Semaphore(threadTotal);
//統計計數結果
final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
//將請求放入執行緒池
for (int i = 0; i < clientTotal ; i++) {
executorService.execute(() -> {
try {
//信號量的獲取
semaphore.acquire();
add();
//釋放
semaphore.release();
} catch (Exception e) {
log.error("exception", e);
}
countDownLatch.countDown();
});
}
countDownLatch.await();
//關閉執行緒池
executorService.shutdown();
log.info("count:{}", count);
}
/**
* 統計方法
*/
private static void add() {
count++;
}
}
運行發現結果隨機,所以非執行緒安全
4 執行緒安全性
4.1 執行緒安全性
當多個執行緒訪問某個類時，不管運行時環境采用何種調度方式或者這些行程將如何交替執行，并且在主調代碼中不需要任何額外的同步或協同，這個類都能表現出正確的行為，那么就稱這個類是執行緒安全的
4.2 原子性
并非一氣呵成，豈能無懈可擊
4.2.1 Atomic 包
- AtomicXXX:CAS,Unsafe.compareAndSwapInt
提供了互斥訪問，同一時刻只能有一個執行緒來對它進行操作
package com.mmall.concurrency.example.atomic;
import com.mmall.concurrency.annoations.ThreadSafe;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;
/**
* @author JavaEdge
*/
@Slf4j
@ThreadSafe
public class AtomicExample2 {
/**
* 請求總數
*/
public static int clientTotal = 5000;
/**
* 同時并發執行的執行緒數
*/
public static int threadTotal = 200;
/**
* 作業記憶體
*/
public static AtomicLong count = new AtomicLong(0);
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(threadTotal);
final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
for (int i = 0; i < clientTotal ; i++) {
executorService.execute(() -> {
try {
System.out.println();
semaphore.acquire();
add();
semaphore.release();
} catch (Exception e) {
log.error("exception", e);
}
countDownLatch.countDown();
});
}
countDownLatch.await();
executorService.shutdown();
//主記憶體
log.info("count:{}", count.get());
}
private static void add() {
count.incrementAndGet();
// count.getAndIncrement();
}
}
@Slf4j
@ThreadSafe
public class AtomicExample4 {
private static AtomicReference<Integer> count = new AtomicReference<>(0);
public static void main(String[] args) {
// 2
count.compareAndSet(0, 2);
// no
count.compareAndSet(0, 1);
// no
count.compareAndSet(1, 3);
// 4
count.compareAndSet(2, 4);
// no
count.compareAndSet(3, 5);
log.info("count:{}", count.get());
}
}
-
AtomicReference,AtomicReferenceFieldUpdater
-
AtomicBoolean
-
AtomicStampReference : CAS的 ABA 問題
4.2.2 鎖
synchronized:依賴 JVM
- 修飾代碼塊:大括號括起來的代碼，作用于呼叫的物件
- 修飾方法: 整個方法，作用于呼叫的物件
- 修飾靜態方法:整個靜態方法，作用于所有物件
package com.mmall.concurrency.example.count;
import com.mmall.concurrency.annoations.ThreadSafe;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
/**
* @author JavaEdge
*/
@Slf4j
@ThreadSafe
public class CountExample3 {
/**
* 請求總數
*/
public static int clientTotal = 5000;
/**
* 同時并發執行的執行緒數
*/
public static int threadTotal = 200;
public static int count = 0;
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(threadTotal);
final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
for (int i = 0; i < clientTotal ; i++) {
executorService.execute(() -> {
try {
semaphore.acquire();
add();
semaphore.release();
} catch (Exception e) {
log.error("exception", e);
}
countDownLatch.countDown();
});
}
countDownLatch.await();
executorService.shutdown();
log.info("count:{}", count);
}
private synchronized static void add() {
count++;
}
}
synchronized 修正計數類方法
- 修飾類:括號括起來的部分，作用于所有物件
子類繼承父類的被 synchronized 修飾方法時,是沒有 synchronized 修飾的!!!
Lock: 依賴特殊的 CPU 指令,代碼實作
4.2.3 對比
- synchronized: 不可中斷鎖，適合競爭不激烈，可讀性好
- Lock: 可中斷鎖，多樣化同步，競爭激烈時能維持常態
- Atomic: 競爭激烈時能維持常態，比Lock性能好; 只能同步一
個值
4.3 可見性
你做的改變，別人看不見，
一個執行緒對主記憶體的修改可以及時的被其他執行緒觀察到
4.3.1 導致共享變數在執行緒間不可見的原因
- 執行緒交叉執行
- 重排序結合執行緒交叉執行
- 共享變數更新后的值沒有在作業記憶體與主存間及時更新
4.3.2 可見性之synchronized
JMM關于synchronized的規定
- 執行緒解鎖前，必須把共享變數的最新值重繪到主記憶體
- 執行緒加鎖時，將清空作業記憶體中共享變數的值，從而使
用共享變數時需要從主記憶體中重新讀取最新的值(加鎖與解鎖是同一把鎖)
4.3.3 可見性之volatile
通過加入記憶體屏障和禁止重排序優化來實作
- 對volatile變數寫操作時，會在寫操作后加入一條store
屏障指令，將本地記憶體中的共享變數值重繪到主記憶體 - 對volatile變數讀操作時，會在讀操作前加入一條load
屏障指令，從主記憶體中讀取共享變數
- volatile使用
volatile boolean inited = false;
//執行緒1:
context = loadContext();
inited= true;
// 執行緒2:
while( !inited ){
sleep();
}
doSomethingWithConfig(context)
4.4 有序性
不按套路出牌，
一個執行緒觀察其他執行緒中的指令執行順序，由于指令重排序的存在，該觀察結果一般雜亂無序
JMM允許編譯器和處理器對指令進行重排序，但是重排序程序不會影響到單執行緒程式的執行，卻會影響到多執行緒并發執行的正確性
4.4.1 happens-before 規則
5發布物件
-
發布物件
使一個物件能夠被當前范圍之外的代碼所使用 -
物件逸出
一種錯誤的發布，當-個物件還沒有構造完成時,就使它被其他執行緒所見
5.1 安全發布物件
package com.mmall.concurrency.example.singleton;
import com.mmall.concurrency.annoations.NotThreadSafe;
/**
* 懶漢模式 -》 雙重同步鎖單例模式
* 單例實體在第一次使用時進行創建
* @author JavaEdge
*/
@NotThreadSafe
public class SingletonExample4 {
/**
* 私有建構式
*/
private SingletonExample4() {
}
// 1、memory = allocate() 分配物件的記憶體空間
// 2、ctorInstance() 初始化物件
// 3、instance = memory 設定instance指向剛分配的記憶體
// JVM和cpu優化，發生了指令重排
// 1、memory = allocate() 分配物件的記憶體空間
// 3、instance = memory 設定instance指向剛分配的記憶體
// 2、ctorInstance() 初始化物件
/**
* 單例物件
*/
private static SingletonExample4 instance = null;
/**
* 靜態的工廠方法
*
* @return
*/
public static SingletonExample4 getInstance() {
// 雙重檢測機制 // B
if (instance == null) {
// 同步鎖
synchronized (SingletonExample4.class) {
if (instance == null) {
// A - 3
instance = new SingletonExample4();
}
}
}
return instance;
}
}
7 AQS
7.1 介紹
- 使用Node實作FIFO佇列，可以用于構建鎖或者其他同步裝置的基礎框架
- 利用了一個int型別表示狀態
- 使用方法是繼承
- 子類通過繼承并通過實作它的方法管理其狀態{acquire 和release} 的方法操縱狀態
- 可以同時實作排它鎖和共享鎖模式(獨占、共享)
同步組件
CountDownLatch
package com.mmall.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author JavaEdge
*/
@Slf4j
public class CountDownLatchExample1 {
private final static int threadCount = 200;
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
final int threadNum = i;
exec.execute(() -> {
try {
test(threadNum);
} catch (Exception e) {
log.error("exception", e);
} finally {
countDownLatch.countDown();
}
});
}
countDownLatch.await();
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws Exception {
Thread.sleep(100);
log.info("{}", threadNum);
Thread.sleep(100);
}
}
package com.mmall.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* 指定時間內處理任務
*
* @author JavaEdge
*
*/
@Slf4j
public class CountDownLatchExample2 {
private final static int threadCount = 200;
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
final int threadNum = i;
exec.execute(() -> {
try {
test(threadNum);
} catch (Exception e) {
log.error("exception", e);
} finally {
countDownLatch.countDown();
}
});
}
countDownLatch.await(10, TimeUnit.MILLISECONDS);
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws Exception {
Thread.sleep(100);
log.info("{}", threadNum);
}
}
##Semaphore用法
##CycliBarrier
package com.mmall.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author JavaEdge
*/
@Slf4j
public class CyclicBarrierExample1 {
private static CyclicBarrier barrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
barrier.await();
log.info("{} continue", threadNum);
}
}
package com.mmall.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author JavaEdge
*/
@Slf4j
public class CyclicBarrierExample2 {
private static CyclicBarrier barrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
try {
barrier.await(2000, TimeUnit.MILLISECONDS);
} catch (Exception e) {
log.warn("BarrierException", e);
}
log.info("{} continue", threadNum);
}
}
package com.mmall.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
/**
* @author JavaEdge
*/
@Slf4j
public class SemaphoreExample3 {
private final static int threadCount = 20;
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < threadCount; i++) {
final int threadNum = i;
exec.execute(() -> {
try {
// 嘗試獲取一個許可
if (semaphore.tryAcquire()) {
test(threadNum);
// 釋放一個許可
semaphore.release();
}
} catch (Exception e) {
log.error("exception", e);
}
});
}
exec.shutdown();
}
private static void test(int threadNum) throws Exception {
log.info("{}", threadNum);
Thread.sleep(1000);
}
}
#9 執行緒池
##9.1 newCachedThreadPool
##9.2 newFixedThreadPool
##9.3 newSingleThreadExecutor
看出是順序執行的
9.4 newScheduledThreadPool
10 死鎖
11 高并發之擴容思路
11.1 擴容
11.1 擴容 - 資料庫
12 高并發之快取思路
12.1 快取
1 快取特征
2 快取命中率影響因素
3 快取分類和應用場景
12.2 高并發之快取-特征、場景及組件介紹
1 Guava Cache
2 快取 - Memchche
3 快取 - Redis
12.3 redis的使用
- 配置類
- 服務類
12.4 高并發場景問題及實戰
快取一致性
13 高并發之訊息佇列思路
13.1 業務案例
將發短信封裝成一條訊息放進訊息佇列中,若發生短信過多,佇列已滿,需要控制發送的頻率.
通過將事件封裝成訊息放入佇列,實作了業務解耦,異步設計,確保了短信服務只要正常后,一定會將短信成功發到用戶.
13.2 訊息佇列的特性
-
為何需要訊息佇列呢
-
優點
佇列
kafka
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/301034.html
標籤：java