Java併發編程核心概念一覽

時間 2019-12-05

標籤 java 併發編程核心概念一覽欄目 Java 简体版

原文原文鏈接

做者博客地址 https://muggle.javaboy.org。java

並行相關概念

同步和異步

同步和異步一般來形容一次方法的調用。同步方法一旦開始，調用者必須等到方法結束才能執行後續動做；異步方法則是在調用該方法後沒必要等到該方法執行完就能執行後面的代碼，該方法會在另外一個線程異步執行，異步方法老是伴隨着回調，經過回調來得到異步方法的執行結果。git

併發和並行

不少人都將併發與並行混淆在一塊兒，它們雖然均可以表示兩個或者多個任務一塊兒執行，但執行過程上是有區別的。併發是多個任務交替執行，多任務之間仍是串行的；而並行是多個任務同時執行，和併發有本質區別。github

對計算機而言，若是系統內只有一個 CPU ，而使用多進程或者多線程執行任務，那麼這種狀況下多線程或者多進程就是併發執行，並行只可能出如今多核系統中。固然，對 Java 程序而言，咱們沒必要去關心程序是並行仍是併發。web

臨界區

臨界區表示的是多個線程共享但同時只能有一個線程使用它的資源。在並行程序中臨界區資源是受保護的，必須確保同一時刻只有一個線程能使用它。redis

阻塞

若是一個線程佔有了臨界區的資源，其餘須要使用這個臨界區資源的線程必須在這個臨界區進行等待(線程被掛起)，這種狀況就是發生了阻塞(線程停滯不前)。spring

死鎖\飢餓\活鎖

死鎖就是多個線程須要其餘線程的資源才能釋放它所擁有的資源，而其餘線程釋放這個線程須要的資源必須先得到這個線程所擁有的資源，這樣形成了矛盾沒法解開；如圖1情形就是發生死鎖現象：數據庫

圖1：生活中的死鎖現象

活鎖就是兩個線程互相謙讓資源，結果就是誰也拿不到資源致使活鎖；就比如過馬路，行人給車讓道，車又給行人讓道，結果就是車和行人都停在那不走。編程

飢餓就是，某個線程優先級特別低總是拿不到資源，致使這個線程一直沒法執行。數組

併發級別

併發級別分爲阻塞，無飢餓，無障礙，無鎖，無等待幾個級別；根據名字咱們也能大概猜出這幾個級別對應的什麼情形；阻塞，無飢餓和無鎖都好理解；咱們說一下無障礙和無等待；緩存

無障礙：無障礙級別默認各個線程不會發生衝突，不會互相搶佔資源，一旦搶佔資源就認爲線程發生錯誤，進行回滾。

無等待：無等待是在無鎖上的進一步優化，限制每一個線程完成任務的步數。

並行的兩個定理

加速比：加速比=優化前系統耗時/優化後系統耗時

Amdahl 定理：加速比=1/[F+(1-F)/n] 其中 n 表示處理器個數，F是程序中只能串行執行的比例(串行率)；由公式可知，想要以最小投入，獲得最高加速比即 F+(1-F)/n 取到最小值，F 和 n 都對結果有很大影響，在深刻研究就是數學問題了。

Gustafson 定律：加速比=n-F(n-1)，這兩定律區別不大，都體現了單純的減小串行率，或者單純的加 CPU 都沒法獲得最優解。

Java 中的並行基礎

原子性，可見性，有序性

原子性指的是一個操做是不可中斷的，要麼成功要麼失敗，不會被其餘線程所幹擾；好比 int=1 ,這一操做在 cpu 中分爲好幾個指令，但對程序而言這幾個指令是一體的，只有可能執行成功或者失敗，不可能發生只執行了一半的操做；對不一樣 CPU 而言保證原子性的的實現方式各有不一樣，就英特爾 CPU 而言是使用一個 lock 指令來保證的。

可見性指某一線程改變某一共享變量，其餘線程未必會立刻知道。

有序性指對一個操做而言指令是按必定順序執行的，但編譯器爲了提升程序執行的速度，會重排程序指令；cpu在執行指令的時候採用的是流水線的形式，上一個指令和下一個指令差一個工步。好比A指令分三個工步：

操做內存a；
操做內存b；
操做內存c；

現假設有個指令 B 操做流程和 A 同樣，那麼先執行指令 A 再執行指令 B 時間全利用上了，中間沒有停頓等待；但若是有三個這樣的指令在流水線上執行： a>b>c ， b>e>c ， c>e>a ；這樣的指令順序就會發生等待下降了 CPU 的效率，編譯器爲了不這種事情發生，會適當優化指令的順序進行重排。

volatile關鍵字

volatile 關鍵字在 Java 中的做用是保證變量的可見性和防止指令重排。

線程的相關操做

建立線程有三種方法

繼承Thread類建立線程
實現Runnable接口建立線程
使用Callable和Future建立線程

終止線程的方法

終止線程可調用 stop() 方法，但這個方法是被廢棄不建議使用的，由於強制終止一個線程會引發數據的不一致問題。好比一個線程數據寫到一半被終止了，釋放了鎖，其餘線程拿到鎖繼續寫數據，結果致使數據發生了錯誤。終止線程比較好的方法是「讓程序本身終止」，好比定義一個標識符，當標識符爲 true 的時候直讓程序走到終點，這樣就能達到「本身終止」的目的。

線程的中斷等待和通知

interrupt() 方法能夠中斷當前程序，object.wait() 方法讓線程進入等待隊列，object.notify() 隨機喚醒等待隊列的一個線程， object.notifyAll() 喚醒等待隊列的全部線程。object.wait() 必須在 synchronzied 語句中調用；執行wait、notify 方法必須得到對象的監視器，執行結束後釋放監視器供其餘線程獲取。

join

join() 方法功能是等待其餘線程「加入」，能夠理解爲將某個線程併爲本身的子線程，等子線程走完或者等子線程走規定的時間，主線程才往下走；join 的本質是調用調用線程對象的 wait 方法，當咱們執行 wait 或者 notify 方法不該該獲取線程對象的監聽器，由於可能會影響到其餘線程的 join。

yield

yield 是線程的「謙讓」機制，能夠理解爲當線程搶到 cpu 資源時，放棄此次資源從新搶佔，yield() 是 Thread 裏的一個靜態方法。

線程組

若是一個多線程系統線程數量衆多並且分工明確，那麼可使用線程組來分類。

public void contextLoads() {
ThreadGroup testGroup=new ThreadGroup("testGroup");
Thread a = new Thread(testGroup, new MyRunnable(), "a");
Thread b = new Thread(testGroup, new MyRunnable(), "b");
a.start();
b.start();
int i = testGroup.activeCount();
}
class MyRunnable implements Runnable{
@Override
public void run() {
System.out.println("test");
}
}

圖示代碼建立了一個 testGroup 線程組。

守護線程

守護線程是一種特殊線程，它相似 Java 中的異常系統，主要是概念上的分類，與之對應的是用戶線程。它功能應該是在後臺完成一些系統性的服務；設置一個線程爲守護線程應該在線程 start 以前 setDaemon()。

線程優先級

Java 中線程能夠有本身的優先級，優先級高的更有優點搶佔資源；線程優先級高的不必定能搶佔到資源，只是一個機率問題，而對應優先級低的線程可能會發生飢餓。

在 Java 中使用1到10表示線程的優先級，使用setPriority()方法來進行設置，數字越大表明優先級越高。

Java 線程鎖

如下分類是從多個同角度來劃分，而不是以某一標準來劃分，請注意：

阻塞鎖：當一個線程得到鎖，其餘線程就會被阻塞掛起，直到搶佔到鎖才繼續執行，這樣會致使 CPU 切換上下文，切換上下文對 CPU 而言是很耗費時間的。
非阻塞鎖：當一個線程得到鎖，其餘線程直接跳過鎖資源相關的代碼繼續執行，就是非阻塞鎖。
自旋鎖：當一個線程得到鎖，其餘線程則在不停進行空循環，直到搶到鎖，這樣作的好處是避免了上下文切換。
可重入鎖：也叫作遞歸鎖，當一個線程得到該鎖後，能夠屢次進入該鎖所同步着的代碼塊。
互斥鎖：互斥鎖保證了某一時刻只能有一個線程佔有該資源。
讀寫鎖：將代碼功能分爲讀和寫，讀不互斥，寫互斥。
公平鎖/非公平鎖：公平鎖就是在等待隊列裏排最前面的的先得到鎖，非公平鎖就是誰搶到誰用。
重量級鎖/輕量級鎖/偏向鎖：使用操做系統「Mutex Lock」功能來實現鎖機制的叫重量級鎖，由於這種鎖成本高；輕量級鎖是對重量級鎖的優化，提升性能；偏向鎖是對輕量級鎖的優化，在無多線程競爭的狀況下儘可能減小沒必要要的輕量級鎖執行路徑。

synchronized

屬於阻塞鎖、互斥鎖、非公平鎖以及可重入鎖，在 JDK1.6 之前屬於重量級鎖，後來作了優化。

用法：

指定加鎖對象
用於靜態代碼塊/方法
用於動態代碼塊/方法

示例：

public static synchronized void test1(){
System.out.println("test");
}
public synchronized void test2(){
System.out.println("test");
}
public void test3(){
synchronized (Main.class){
System.out.println("test");
}
}

當鎖加在靜態代碼塊上或者靜態方法上或者爲 synchronized(xxx.class){} 時，鎖做用於整個類，凡是屬於這個類的對象的相關都會被上鎖，當用於動態方法或者爲或者爲synchronized (object){}時鎖做用於對象；除此以外，synchronized能夠保證線程的可見性和有序性。

Lock

Lock 是一個接口，其下有多個實現類。

方法說明：

lock()方法是日常使用得最多的一個方法，就是用來獲取鎖。若是鎖已被其餘線程獲取，則進行等待。
tryLock()方法是有返回值的，它表示用來嘗試獲取鎖，若是獲取成功，則返回true，若是獲取失敗（即鎖已被其餘線程獲取），則返回false，這個方法還能夠設置一個獲取鎖的等待時長，若是時間內獲取不到直接返回。
兩個線程同時經過lock.lockInterruptibly()想獲取某個鎖時，倘若此時線程A獲取到了鎖，而線程B只有在等待，那麼對線程B調用threadB.interrupt()方法可以中斷線程B的等待過程。
unLock()方法是用來釋放鎖。
newCondition()：生成一個和線程綁定的Condition實例，利用該實例咱們可讓線程在合適的時候等待，在特定的時候繼續執行，至關於獲得這個線程的wait和notify方法。

ReentrantLock

ReentrantLock 重入鎖，是實現 Lock 接口的一個類，它對公平鎖和非公平鎖都支持，在構造方法中傳入一個 boolean 值，true 時爲公平鎖，false 時爲非公平鎖。

Semaphore(信號量)

信號量是對鎖的擴展，鎖每次只容許一個線程訪問一個資源，而信號量卻能夠指定多個線程訪問某個資源，信號量的構造函數爲

public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

第一個方法指定了可以使用的線程數，第二個方法的布爾值表示是否爲公平鎖。

acquire() 方法嘗試得到一個許可，若是獲取不到則等待；tryAcquire() 方法嘗試獲取一個許可，成功返回 true，失敗返回false，不會阻塞，tryAcquire(int i) 指定等待時間；release() 方法釋放一個許可。

ReadWriteLock

讀寫分離鎖，讀寫分離鎖能夠有效的減小鎖競爭，讀鎖是共享鎖，能夠被多個線程同時獲取，寫鎖是互斥只能被一個線程佔有，ReadWriteLock 是一個接口，其中 readLock() 得到讀鎖，writeLock() 得到寫鎖其實現類 ReentrantReadWriteLock 是一個可重入得的讀寫鎖，它支持鎖的降級(在得到寫鎖的狀況下能夠再持有讀鎖)，不支持鎖的升級（在得到讀鎖的狀況下不能再得到寫鎖）；讀鎖和寫鎖也是互斥的，也就是一個資源要麼被上了一個寫鎖，要麼被上了多個讀鎖，不會發生這個資即被上寫鎖又被上讀鎖的狀況。

cas

cas(比較替換)：無鎖策略的一種實現方式，過程爲獲取到變量舊值（每一個線程都有一份變量值的副本），和變量目前的新值作比較，若是同樣證實變量沒被其餘線程修改過，這個線程就能夠更新這個變量，不然不能更新；通俗的說就是經過不加鎖的方式來修改共享資源並同時保證安全性。

使用cas的話對於屬性變量不能再用傳統的 int ,long 等；要使用原子類代替原先的數據類型操做，好比 AtomicBoolean，AtomicInteger，AtomicInteger 等。

併發下集合類

併發集合類主要有：

ConcurrentHashMap：支持多線程的分段哈希表，它經過將整個哈希表分紅多段的方式減少鎖粒度。
ConcurrentSkipListMap：ConcurrentSkipListMap的底層是經過跳錶來實現的。跳錶是一個鏈表，可是經過使用「跳躍式」查找的方式使得插入、讀取數據時複雜度變成了O（logn）。
ConCurrentSkipListSet：參考 ConcurrentSkipListMap。
CopyOnWriteArrayList：是 ArrayList 的一個線程安全的變形，其中全部可變操做（添加、設置，等等）都是經過對基礎數組進行一次新的複製來實現的。
CopyOnWriteArraySet：參考 CopyOnWriteArrayList。
ConcurrentLinkedQueue：cas 實現的非阻塞併發隊列。

線程池

介紹

多線程的設計優勢是能很大限度的發揮多核處理器的計算能力，可是，若不控制好線程資源反而會拖累cpu，下降系統性能，這就涉及到了線程的回收複用等一系列問題；並且自己線程的建立和銷燬也很耗費資源，所以找到一個合適的方法來提升線程的複用就很必要了。

線程池就是解決這類問題的一個很好的方法：線程池中自己有不少個線程，當須要使用線程的時候拿一個線程出來，當用完則還回去，而不是每次都建立和銷燬。在 JDK 中提供了一套 Executor 線程池框架，幫助開發人員有效的進行線程控制。

Executor 使用

得到線程池的方法：

newFixedThreadPool(int nThreads) ：建立固定數目線程的線程池。
newCachedThreadPool：建立一個可緩存的線程池，調用execute將重用之前構造的線程（若是線程可用）。若是現有線程沒有可用的，則建立一個新線程並添加到池中。
newSingleThreadExecutor：建立一個單線程化的 Executor。
newScheduledThreadPool：建立一個支持定時及週期性的任務執行的線程池。

以上方法都是返回一個 ExecutorService 對象，executorService.execute() 傳入一個 Runnable 對象，可執行一個線程任務。

下面看示例代碼

public class Test implements Runnable{
int i=0;
public Test(int i){
this.i=i;
}
public void run() {
System.out.println(Thread.currentThread().getName()+"====="+i);
}
public static void main(String[] args) throws InterruptedException {
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for(int i=0;i<10;i++){
cachedThreadPool.execute(new Test(i));
Thread.sleep(1000);
}
}
}

線程池是一個龐大而複雜的體系，本文定位是基礎，不對其作更深刻的研究，感興趣的小夥伴能夠自行查資料進行學習。

ScheduledExecutorService

newScheduledThreadPool(int corePoolSize) 會返回一個ScheduledExecutorService 對象，能夠根據時間對線程進行調度；其下有三個執行線程任務的方法：schedule()，scheduleAtFixedRate() 以及 scheduleWithFixedDelay() 該線程池可解決定時任務的問題。

示例：

class Test implements Runnable {
private String testStr;
Test(String testStr) {
this.testStr = testStr;
}
@Override
public void run() {
System.out.println(testStr + " >>>> print");
}
public static void main(String[] args) {
ScheduledExecutorService service = Executors.newScheduledThreadPool(10);
long wait = 1;
long period = 1;
service.scheduleAtFixedRate(new MyScheduledExecutor("job1"), wait, period, TimeUnit.SECONDS);
service.scheduleWithFixedDelay(new MyScheduledExecutor("job2"), wait, period, TimeUnit.SECONDS);
scheduledExecutorService.schedule(new MyScheduledExecutor("job3"), wait, TimeUnit.SECONDS);//延時waits 執行
}
}

job1的執行方式是任務發起後間隔 wait 秒開始執行，每隔 period 秒(注意：不包括上一個線程的執行時間)執行一次；

job2的執行方式是任務發起後間隔 wait 秒開始執行，等線程結束後隔 period 秒開始執行下一個線程；

job3只執行一次，延遲 wait 秒執行；

ScheduledExecutorService 還能夠配合 Callable 使用來回調得到線程執行結果，還能夠取消隊列中的執行任務等操做，這屬於比較複雜的用法，咱們這裏掌握基本的便可，到實際遇到相應的問題時咱們在現學現用，節省學習成本。

鎖優化

減少鎖持有時間

減少鎖的持有時間可有效的減小鎖的競爭。若是線程持有鎖的時間越長，那麼鎖的競爭程度就會越激烈。所以，應儘量減小線程對某個鎖的佔有時間，進而減小線程間互斥的可能。

減小鎖持有時間的方法有：

進行條件判斷，只對必要的狀況進行加鎖，而不是整個方法加鎖。
減小加鎖代碼的行數，只對必要的步驟加鎖。

減少鎖粒度

減少鎖的範圍，減小鎖住的代碼行數可減小鎖範圍，減少共享資源的範圍也可減少鎖的範圍。減少鎖共享資源的範圍的方式比較常見的有分段鎖，好比 ConcurrentHashMap ，它將數據分爲了多段，當須要 put 元素的時候，並非對整個 hashmap 進行加鎖，而是先經過 hashcode 來知道他要放在那一個分段中，而後對這個分段進行加鎖，因此當多線程 put 的時候，只要不是放在一個分段中，就實現了真正的並行的插入。

鎖分離

鎖分離最多見的操做就是讀寫分離了，讀寫分離的操做參考 ReadWriteLock 章節，而對讀寫分離進一步的延伸就是鎖分離了。爲了提升線程的並行量，咱們能夠針對不一樣的功能採用不一樣的鎖，而不是通通用同一把鎖。好比說有一個同步方法未進行鎖分離以前，它只有一把鎖，任何線程來了，只有拿到鎖纔有資格運行，進行鎖分離以後就不是這種情形了——來一個線程，先判斷一下它要幹嗎，而後發一個對應的鎖給它，這樣就能必定程度上提升線程的並行數。

鎖粗化

通常爲了保證多線程間的有效併發，會要求每一個線程持有鎖的時間儘可能短，也就是說鎖住的代碼儘可能少。可是若是若是對同一個鎖不停的進行請求、同步和釋放，其自己也會消耗系統寶貴的資源，反而不利於性能的優化。好比有三個步驟：a、b、c，a同步，b不一樣步，c同步；那麼一個線程來時候會上鎖釋放鎖而後又上鎖釋放鎖。這樣反而可能會下降線程的執行效率，這個時候咱們將鎖粗化可能會更好——執行 a 的時候上鎖，執行完 c 再釋放鎖。

鎖擴展

分佈式鎖

JDK 提供的鎖在單體項目中不會有什麼問題，可是在集羣項目中就會有問題了。在分佈式模型下，數據只有一份（或有限制），此時須要利用鎖的技術控制某一時刻修改數據的進程數。JDK 鎖顯然沒法知足咱們的需求，因而就有了分佈式鎖。

分佈式鎖的實現有三種方式：

基於數據庫實現分佈式鎖
基於緩存（redis，memcached，tair）實現分佈式鎖
基於 Zookeeper 實現分佈式鎖

基於redis的分佈式鎖比較使用廣泛，在這裏介紹其原理和使用：

redis 實現鎖的機制是 setnx 指令，setnx 是原子操做命令，鎖存在不能設置值，返回 0 ；鎖不存在，則設置鎖，返回 1 ，根據返回值來判斷上鎖是否成功。看到這裏你可能想爲啥不先 get 有沒有值，再 set 上鎖；首先咱們要知道，redis 是單線程的，同一時刻只可能有一個線程操做內存，而後 setnx 是一個操做步驟(具備原子性)，而 get 再 set 是兩個步驟（不具備原子性）。若是使用第二種可能會發生這種狀況：客戶端 a get發現沒有鎖，這個時候被切換到客戶端b，b get也發現沒鎖，而後b set，這個時候又切換到a客戶端 a set；這種狀況下，鎖徹底沒起做用。因此，redis分佈式鎖，原子性是關鍵。

對於 web 應用中 redis 客戶端用的比較多的是 lettuce，jedis，redisson。springboot 的 redis 的 start 包底層是 lettuce ，但對 redis 分佈式鎖支持得最好的是 redisson（若是用 redisson 你就享受不到 redis 自動化配置的好處了）；不過 springboot 的 redisTemplete 支持手寫 lua 腳本，咱們能夠經過手寫 lua 腳原本實現 redis 鎖。

代碼示例：

public boolean lockByLua(String key, String value, Long expiredTime){
String strExprie = String.valueOf(expiredTime);
StringBuilder sb = new StringBuilder();
sb.append("if redis.call(\"setnx\",KEYS[1],ARGV[1])==1 ");
sb.append("then ");
sb.append(" redis.call(\"pexpire\",KEYS[1],KEYS[2]) ");
sb.append(" return 1 ");
sb.append("else ");
sb.append(" return 0 ");
sb.append("end ");
String script = sb.toString();
RedisCallback<Boolean> callback = (connection) -> {
return connection.eval(script.getBytes(), ReturnType.BOOLEAN, 2, key.getBytes(Charset.forName("UTF-8")),strExprie.getBytes(Charset.forName("UTF-8")), value.getBytes(Charset.forName("UTF-8")));
};
Boolean execute = stringRedisTemplate.execute(callback);
return execute;
}

關於lua腳本的語法我就不作介紹了。

在 github 上也有開源的 redis 鎖項目，好比 spring-boot-klock-starter 感興趣的小夥伴能夠去試用一下。

數據庫鎖

對於存在多線程問題的項目，好比商品貨物的進銷存，訂單系統單據流轉這種，咱們能夠經過代碼上鎖來控制併發，也可使用數據庫鎖來控制併發，數據庫鎖從機制上來講分樂觀鎖和悲觀鎖。

悲觀鎖：

悲觀鎖分爲共享鎖（S鎖）和排他鎖（X鎖），MySQL 數據庫讀操做分爲三種——快照讀，當前讀；快照讀就是普通的讀操做，如：

select *from table

當前讀就是對數據庫上悲觀鎖了；其中 select...lockinshare mode 屬於共享鎖，多個事務對於同一數據能夠共享，但只能讀不能修改。而下面三種 SQL :

select ...for update
update ... set...
insert into ...

屬於排他鎖，排他鎖就是不能與其餘鎖並存，如一個事務獲取了一個數據行的排他鎖，其餘事務就不能再獲取該行的其餘鎖，包括共享鎖和排他鎖，可是獲取排他鎖的事務是能夠對數據行讀取和修改，排他鎖是阻塞鎖。

樂觀鎖：

就是很樂觀，每次去拿數據的時候都認爲別人不會修改，因此不會上鎖，可是在更新的時候會判斷一下在此期間別人有沒有去更新這個數據，若是有則更新失敗。一種實現方式爲在數據庫表中加一個版本號字段 version ，任何 update 語句 where 後面都要跟上 version=？，而且每次 update 版本號都加 1。若是 a 線程要修改某條數據，它須要先 select 快照讀得到版本號，而後 update ，同時版本號加一。這樣就保證了在 a 線程修改某條數據的時候，確保其餘線程沒有修改過這條數據，一旦其餘線程修改過，就會致使 a 線程版本號對不上而更新失敗（這實際上是一個簡化版的mvcc）。

樂觀鎖適用於容許更新失敗的業務場景，悲觀鎖適用於確保更新操做被執行的場景。

併發編程相關

善用 Java8 Stream
對於生產者消費者模式，條件判斷是使用 while 而不是 if
懶漢單例採用雙重檢查和鎖保證線程安全
善用 Future 模式
合理使用 ThreadLocal

Java 8 引入 lambda 表達式使在 Java 中使用函數式編程很方便。而 Java 8 中的 stream 對數據的處理能使線程執行速度得以優化。Future 模式是一種對異步線程的回調機制；如今 cpu 都是多核的，咱們在處理一些較爲費時的任務時可以使用異步，在後臺開啓多個線程同時處理，等到異步線程處理完再經過 Future 回調拿處處理的結果。

ThreadLocal 的實例表明了一個線程局部的變量，每條線程都只能看到本身的值，並不會意識到其它的線程中也存在該變量(這裏原理就不說了，網上資料不少)，總之就是咱們若是想在多線程的類裏面使用線程安全的變量就用 ThreadLocal ，可是請必定要注意用完記得 remove ，否則會發生內存泄漏。