Java并發(fā)編程（JUC）模擬AND型信號量
AND型信號量可能大家都聽說過并可能都有一定的理解，但是你有使用過么?今天就使用Java來模擬實(shí)現(xiàn)!
本文是對上篇文章(進(jìn)程同步機(jī)制)的一次實(shí)踐，通過JUC提供的一些機(jī)制來模擬一些OS中的AND型信號量，因?yàn)橛涗浶托托盘柫靠梢缘葍r(jià)于JUC中提供的Semaphore(信號量)，但是對于AND型信號量因?yàn)橐恍┰?主要是過時(shí)了)，JUC沒有提供，今天就手動(dòng)的來寫一個(gè)AND型信號量對應(yīng)的Swait操作和Ssignal操作(這里不明白的可以看前面的理論篇)。通過本篇博文讓你對進(jìn)程同步機(jī)制有個(gè)更好的理解。
1.一個(gè)錯(cuò)誤示例
在這里，首先解釋一下，為了滿足線程申請信號量不成功后將進(jìn)程阻塞，并插入到對應(yīng)的隊(duì)列中，所以使用了ReentrantLock+Condition來實(shí)現(xiàn)Swait方法。廢話不多說，直接上代碼：
- //數(shù)據(jù)定義
- static Lock lock = new ReentrantLock();
- static Condition condition1 = lock.newCondition();
- static Condition condition2 = lock.newCondition();
- public static void Swait(String id, Semaphore s1, Semaphore s2) throws InterruptedException {
- lock.tryLock(1, TimeUnit.SECONDS);
- log.info("當(dāng)前的兩個(gè)信號量的狀態(tài)：【{},{}】", s1.availablePermits(), s2.availablePermits());
- //availablePermits可獲取到信號量中還剩余的值
- if(s1.availablePermits() < 1 || s2.availablePermits() < 1){
- if (s1.availablePermits() < 1) {
- log.info("線程【{}】被掛起到信號量【{}】中", id, s1);
- //阻塞，并插入到condition1的阻塞隊(duì)列中
- condition1.await();
- } else {
- log.info("線程【{}】被掛起到信號量【{}】中", id, s2);
- //阻塞，并插入到condition2的阻塞隊(duì)列中
- condition2.await();
- }
- log.info("被掛起的線程【{}】被喚醒執(zhí)行。", id);
- } else {
- log.info("為線程【{}】分配資源！", id);
- s1.acquire();
- s2.acquire();
- }
- lock.unlock();
- }
- public static void Ssignal(Semaphore s1, Semaphore s2) throws InterruptedException {
- log.info("線程【{}】執(zhí)行了釋放資源", id);
- lock.tryLock(1, TimeUnit.SECONDS);
- s1.release();
- s2.release();
- //喚醒等待隊(duì)列中的線程
- condition.signal();
- lock.unlock();
- }
大家仔細(xì)看上面的代碼，這個(gè)也是我剛開始寫的代碼，第一眼看似乎是沒什么問題，但是里面隱藏著一個(gè)坑，在Swait方法中，調(diào)用condition1.await()，此時(shí)線程被阻塞在這一行中，但是當(dāng)被別的線程(調(diào)用Ssignal)喚醒時(shí)，在被阻塞的下一行開始繼續(xù)執(zhí)行，但是在后續(xù)的代碼里，是沒有去申請信號量的，而是直接就Swait成功了，這樣在執(zhí)行Ssignal時(shí)就會(huì)導(dǎo)致信號量憑空的增加了，也就無法正確的表征系統(tǒng)中的資源數(shù)量了。
2.一個(gè)簡單的示例
下面我們就對代碼進(jìn)行優(yōu)化，大家可以回顧一下AND型信號量，當(dāng)其因?yàn)橘Y源不足時(shí)，需要將線程插入到第一個(gè)無法滿足條件(即Si<1)的信號量對應(yīng)的等待隊(duì)列中，并且將程序計(jì)數(shù)器放置到Swait操作的開始處，所以我們對Swait代碼進(jìn)行修改如下：
- public static void Swait(String id, Semaphore s1, Semaphore s2) throws InterruptedException {
- lock.tryLock(1, TimeUnit.SECONDS);
- log.info("當(dāng)前的兩個(gè)信號量的狀態(tài)：【{},{}】", s1.availablePermits(), s2.availablePermits());
- //如果申請不到，就掛起線程，并將線程插入到condition的隊(duì)列中
- while (s1.availablePermits() < 1 || s2.availablePermits() < 1) {
- if (s1.availablePermits() < 1) {
- log.info("線程【{}】被掛起到信號量【{}】中", id, s1);
- condition1.await();
- } else {
- log.info("線程【{}】被掛起到信號量【{}】中", id, s2);
- condition2.await();
- }
- log.info("被掛起的線程【{}】被喚醒執(zhí)行。", id);
- }
- log.info("為線程【{}】分配資源！", id);
- s1.acquire();
- s2.acquire();
- lock.unlock();
- }
在上面的代碼中，我們將請求的資源放到一個(gè)循環(huán)條件中，以滿足將程序計(jì)數(shù)器放置到Swait操作的開始處，在每次被喚醒后都要重新判斷資源是否足夠，如果足夠才跳出循環(huán)，否則就再次自我阻塞。
3.一個(gè)可以同時(shí)申請N個(gè)的Swait操作
如果你知道了信號量的種類數(shù)(系統(tǒng)中的資源類型)，其實(shí)上面的代碼已經(jīng)可以滿足一定的需要了，只需要我們將所有的信號量寫入到參數(shù)列表中即可。但是對于致力于代碼的復(fù)用，這里就有些差強(qiáng)人意了，因此我們再次對代碼進(jìn)行改進(jìn)，代碼如下所示：
- public static void Swait(String id, Semaphore... list) throws InterruptedException {
- lock.lock();
- //如果資源不足，就掛起線程，并將線程插入到condition的隊(duì)列中
- while (true) {
- int count=0;
- //循環(huán)判斷參數(shù)列表中信號量的可用值
- for (Semaphore semaphore:list){
- if(semaphore.availablePermits()>0){
- count++;
- }
- }
- //如果資源都滿足，則跳出循環(huán)，進(jìn)行資源分配
- if(count == list.length){
- break;
- }
- log.info("線程【{}】被掛起-----", id);
- //將當(dāng)前線程阻塞
- condition1.await();
- log.info("被掛起的線程【{}】被喚醒執(zhí)行。", id);
- }
- log.info("為線程【{}】分配資源！", id);
- //分配資源
- for (Semaphore semaphore:list){
- semaphore.acquire();
- }
- lock.unlock();
- }
- public static void Ssignal(String id, Semaphore... list) throws InterruptedException {
- log.info("線程【{}】執(zhí)行了釋放資源", id);
- lock.tryLock(1, TimeUnit.SECONDS);
- //循環(huán)釋放信號量
- for (Semaphore semaphore:list){
- semaphore.release();
- }
- //喚醒等待隊(duì)列中的線程
- condition.signal();
- lock.unlock();
- }
為此，我們將方法中的信號量列表改為可變的參數(shù)列表，這樣在傳參的時(shí)候就可以方便的進(jìn)行了，但是也會(huì)存才一些問題，比如無法約束“借出”與“歸還”的信號量的數(shù)量是否一致。并且因?yàn)樾盘柫康臄?shù)量不定，所以無法為每個(gè)信號量新建一個(gè)條件變量(Condition)，因此在上面的代碼中所有的信號量公用一個(gè)條件變量，所有阻塞的線程都插入在其阻塞隊(duì)列中。
4.一個(gè)完整的例子
這里我們使用一個(gè)經(jīng)典的進(jìn)程同步問題來演示我們使用Java模擬的AND型信號量，在這里，我們采用生產(chǎn)者–消費(fèi)者問題來演示，完整的代碼如下：
- //用來保證互斥的訪問臨界區(qū)(緩存區(qū))
- static final Semaphore mutex = new Semaphore(1);
- //緩沖區(qū)，最大容量為50
- static List<Integer> buffer = new ArrayList<>();
- //緩沖區(qū)中還可放入的消息數(shù)量
- static final Semaphore empty = new Semaphore(50);
- //緩沖區(qū)中的消息數(shù)量
- static final Semaphore full = new Semaphore(0);
- //可重入鎖和條件變量
- static Lock lock = new ReentrantLock();
- static Condition condition = lock.newCondition();
- //用與輔助的簡單的生成消息
- static Integer count = 0;
- //生產(chǎn)者
- static class Producer extends Thread {
- Producer(String name) {
- super.setName(name);
- }
- @Override
- public void run() {
- do {
- try {
- Swait(this.getName(), mutex, empty);
- log.info("生產(chǎn)了一條消息：【{}】", count);
- buffer.add(count++);
- Thread.sleep(1000);
- Ssignal(this.getName(), mutex, full);
- } catch (InterruptedException e) {
- log.error("生產(chǎn)消息時(shí)產(chǎn)生異常！");
- }
- } while (true);
- }
- }
- //消費(fèi)者
- static class Consumer extends Thread {
- Consumer(String name) {
- super.setName(name);
- }
- @Override
- public void run() {
- do {
- try {
- Swait(this.getName(), mutex, full);
- log.info("消費(fèi)了一條消息：【{}】", buffer.remove(0));
- Thread.sleep(1000);
- Ssignal(this.getName(), mutex, empty);
- } catch (InterruptedException e) {
- log.error("消費(fèi)消息時(shí)產(chǎn)生異常！");
- }
- } while (true);
- }
- }
- public static void Swait(String id, Semaphore... list) throws InterruptedException {
- lock.lock();
- //如果資源不足，就掛起線程，并將線程插入到condition的隊(duì)列中
- while (true) {
- int count=0;
- for (Semaphore semaphore:list){
- if(semaphore.availablePermits()>0){
- count++;
- }
- }
- if(count == list.length){
- break;
- }
- log.info("線程【{}】被掛起", id);
- condition.await();
- log.info("被掛起的線程【{}】被喚醒執(zhí)行。", id);
- }
- log.info("為線程【{}】分配資源！", id);
- for (Semaphore semaphore:list){
- semaphore.acquire();
- }
- lock.unlock();
- }
- public static void Ssignal(String id, Semaphore... list) throws InterruptedException {
- log.info("線程【{}】執(zhí)行了釋放資源", id);
- lock.tryLock(1, TimeUnit.SECONDS);
- for (Semaphore semaphore:list){
- semaphore.release();
- }
- //喚醒等待隊(duì)列中的一個(gè)線程
- condition.signal();
- lock.unlock();
- }
- public static void main(String[] args) {
- Producer p1 = new Producer("p1");
- Consumer c1 = new Consumer("c1");
- p1.start();
- c1.start();
- }
上面代碼都是可以直接執(zhí)行的，如果不需要使用參數(shù)列表，可以將上面的Swait方法進(jìn)行替換即可(記得創(chuàng)建對應(yīng)的條件變量)。
下圖是部分的執(zhí)行結(jié)果：
又到了分隔線以下，本文到此就結(jié)束了，本文內(nèi)容全部都是由博主自己進(jìn)行整理并結(jié)合自身的理解并且進(jìn)行的代碼編寫，如果有什么錯(cuò)誤，還請批評指正。
本文的所有java代碼都已通過測試，對其中有什么疑惑的，可以評論區(qū)留言，歡迎你的留言與討論;另外原創(chuàng)不易，如果本文對你有所幫助，還請留下個(gè)贊，以表支持。
希望本文可以幫助你理解加深理解進(jìn)程同步，也可以幫助你理解Java并發(fā)編程.
