Basic Of Concurrency(十七: Semaphores)

Semaphores是一個用於保證線程間相互發送信號且信號不會丟失的同步結構,與Lock相似,一樣可以保證臨界區的安全訪問.在Java5java.util.concurrent包中已有相關實現,所以咱們不須要本身手動實現.但仍然頗有有必要知道怎麼使用它們以及底層原理.

簡單的Semaphore

如下是一個簡單的Semaphores實現:

public class Semaphore {
    private boolean signal = false;

    public synchronized void take() {
        signal = true;
        notify();
    }

    public synchronized void release() {
        while (!signal) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        signal = false;
    }
}
複製代碼

在Semaphore中,調用take()發送信號會將信號先存儲起來.而調用release()則是等待信號.當收到信號時會從新清除信號並退出release()方法.

像這樣調用Semaphore可以幫助咱們解決信號丟失問題.你可使用take()和release()來代替wait()和notify()方法.若是在調用release()以前就已經調用了take(),再去調用release()仍然可以知道take()被調用過,由於take()調用會將信號存儲在signal變量中.而不像wait()和notify()那樣.

take()和release()方法的命名是有淵源的.它們的名字來源於將semaphores當成鎖來用的時候.稍後會做解釋.

使用Semaphore通信

這是一個簡單的實例,兩個線程使用Semaphore互相通信.

public class SemaphoreDemo {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore();
        Sender sender = new Sender(semaphore);
        Receiver receiver = new Receiver(semaphore);
    }
}
複製代碼

public class Sender extends Thread{
    private Semaphore semaphore;

    public Sender(Semaphore semaphore){
        this.semaphore = semaphore;
    }

    @Override
    public void run(){
        while (true){
            // do something then send signal
            semaphore.take();
        }
    }
}
複製代碼

public class Receiver extends Thread {
    private Semaphore semaphore;

    public Receiver(Semaphore semaphore) {
        this.semaphore = semaphore;
    }

    @Override
    public void run() {
        while (true) {
            semaphore.release();
            // receive signal then do something
        }
    }
}
複製代碼

CountingSemaphore

在前文給出的Semaphore實現並無記錄take()調用了多少次即發送了多少個信號.咱們能夠作出小小的改動來實現這個功能.咱們稱這種實現爲CountingSemaphore:

public class CountingSemaphore {
    private int signal = 0;

    public synchronized void take() {
        signal++;
        notify();
    }

    public synchronized void release() {
        while (signal == 0) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        signal--;
    }
}
複製代碼

BoundedSemaphore

CountingSemaphore並無設置它所能存儲信號數量的上限.咱們能夠作出小小改動來實現這個功能.以下:

public class BoundedSemaphore {
    private int signal = 0;
    private int bound;

    public BoundedSemaphore(int upperBound) {
        this.bound = upperBound;
    }

    public synchronized void take() {
        while (signal == bound) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        signal++;
        notify();
    }

    public synchronized void release() {
        while (signal == 0) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        signal--;
        notify();
    }
}
複製代碼

咱們能夠注意到在信號達到上限時,調用take()會發生阻塞進入等待狀態.BoundedSemaphore在到達信號上限時會阻塞take()調用直到有線程調用release()減小信號數量爲止.

將BoundedSemaphore當成鎖來使用

咱們可以將BoundedSemaphore當成鎖來使用.咱們只須要將信號上限設置爲1,將臨界區代碼經過take()和release()調用包裹起來便可.下面是一個示例:

BoundedSemaphore semaphore = new BoundedSemaphore(1);
semaphore.take();
try{
  //critical section
} finally {
  semaphore.release();
}
複製代碼

上面示例中,take()和release()的調用在同一個線程中完成.當一個線程取得Semaphore實例時,其餘線程將會被阻塞在take()調用上,直到持有Semaphore實例的線程調用release()爲止.當線程調用take()退出後能夠正常調用release()而不會進入阻塞.

此外你還可使用BoundedSemaphore的上限來限制同時進入臨界區的線程數量.在上面實例中,咱們能夠限制BoundedSemaphore的上限爲5.那麼5個線程能夠同時訪問臨界區代碼,前提是這5個線程的操做不會有線程安全問題,否則你的應用將會產生不可預期的後果.

咱們將release()調用放置在finally語句裏面,用於避免臨界區代碼拋出異常而不能釋放信號.

該系列博文爲筆者複習基礎所著譯文或理解後的產物,複習原文來自Jakob Jenkov所著Java Concurrency and Multithreading Tutorial

上一篇: Reentrance Lockout
下一篇: 阻塞隊列