Thread和goroutine兩種方式怎樣實(shí)現(xiàn)共享變量按序輸出

背景

最近在看go的一些底層實(shí)現(xiàn)，其中印象最為深刻的是go語言創(chuàng)造者之一Rob Pike說過的一句話，不要通過共享內(nèi)存通信，而應(yīng)該通過通信來共享內(nèi)存，其中這后半句話對(duì)應(yīng)的實(shí)現(xiàn)是通道（channel），利用通道在多個(gè)協(xié)程（goroutine）之間傳遞數(shù)據(jù)?？吹竭@里，我不禁產(chǎn)生了一個(gè)疑問，對(duì)于無狀態(tài)數(shù)據(jù)之間的傳遞，通過通道保證數(shù)據(jù)之間并發(fā)安全沒什么問題，但我現(xiàn)在有一個(gè)臨界區(qū)或者共享變量，存在多線程并發(fā)訪問。Go協(xié)程如何控制數(shù)據(jù)并發(fā)安全性？難道還有其它高招？帶著這個(gè)疑問，我們看看Go是如何保證臨界區(qū)共享變量并發(fā)訪問問題。

下面我們通過一個(gè)經(jīng)典的題目來驗(yàn)證線程和協(xié)程分別是如何解決的。

有三個(gè)線程/協(xié)程完成如下任務(wù)：1線程/協(xié)程打印1，2線程/協(xié)程打印2，3線程/協(xié)程打印3，依次交替打印15次。輸出：123123123123123

java實(shí)現(xiàn)

java對(duì)于這個(gè)問題如何解決呢？首先要求依次輸出，那么只要保證線程互相等待或者說步調(diào)一致即可實(shí)現(xiàn)上述問題。

如何實(shí)現(xiàn)步調(diào)一致呢？我知道的方法至少有三種，以下我通過三種實(shí)現(xiàn)方式來介紹Java線程是如何控制臨界區(qū)共享變量并發(fā)訪問。

  Synchronized實(shí)現(xiàn) 

通過Synchronized解決互斥問題; （wait/notifyAll）等待-通知機(jī)制控制多個(gè)線程之間執(zhí)行節(jié)奏。實(shí)現(xiàn)方式如下：

public class Thread123 {

 public static void main(String[] args) throws InterruptedException {
  Thread123 testABC = new Thread123();

   Thread thread1 = new Thread(new Runnable() {
    @Override
    public void run() {
     try {
      for (int i = 0; i < 5; i++) {
       testABC.printA();
      }
     } catch (InterruptedException e) {
      e.printStackTrace();
     }
    }
   });
   Thread thread2 = new Thread(new Runnable() {
    @Override
    public void run() {
     try {
      for (int i = 0; i < 5; i++) {
       testABC.printB();
      }
     } catch (InterruptedException e) {
      e.printStackTrace();
     }
    }
   });
   Thread thread3 = new Thread(new Runnable() {
    @Override
    public void run() {
     try {
      for (int i = 0; i < 5; i++) {
       testABC.printC();
      }
     } catch (InterruptedException e) {
      e.printStackTrace();
     }
    }
   });
   thread1.start();
   thread2.start();
   thread3.start();
   thread1.join();
   thread2.join();
   thread3.join();
 }
 int flag = 1;
 public synchronized void printA() throws InterruptedException {
  while (flag != 1) {
   this.wait();
  }
  System.out.print(flag);
  flag = 2;
  this.notifyAll();
 }

 private synchronized void printB() throws InterruptedException {
  while (flag != 2) {
   this.wait();
  }
  System.out.print(flag);
  flag = 3;
  this.notifyAll();
 }

 private synchronized void printC() throws InterruptedException {
  while (flag != 3) {
   this.wait();
  }
  System.out.print(flag);
  flag = 1;
  this.notifyAll();
 }
}

看到這段實(shí)現(xiàn)可能大家都會(huì)有如下兩個(gè)疑問：

• 為啥要用notifyAll，而沒有使用notify？

“  

這兩者其實(shí)是有一定區(qū)別的，notify是隨機(jī)的通知等待隊(duì)列中的一個(gè)線程，而notifyAll是通知等待隊(duì)列中所有的線程?？赡芪覀兊谝桓杏X是即使使用了notifyAll也是只能有一個(gè)線程真正執(zhí)行，但是在多線程編程中，所謂的感覺都蘊(yùn)藏著風(fēng)險(xiǎn)，因?yàn)橛行┚€程可能永遠(yuǎn)也不會(huì)被喚醒，這就導(dǎo)致即使?jié)M足條件也無法執(zhí)行，所以除非你很清楚你的線程執(zhí)行邏輯，一般情況下，不要使用notify。有興趣的話，上面例子，可以測試下，你就可以得知為什么不建議你用notify。

”  

• 為啥要用while循環(huán)，而不是用更輕量的if？

“  

利用while的原因，從根本上來說是java中的編程范式，只要涉及到wait等待，都需要用while。原因是因?yàn)楫?dāng)wait返回時(shí)，有可能判斷條件已經(jīng)發(fā)生變化，所以需要重新檢驗(yàn)條件是否滿足。

”  

Lock實(shí)現(xiàn)

通過Lock解決多線程之間互斥問題; (await/signal)解決線程之間同步，當(dāng)然這種實(shí)現(xiàn)方式和上一種效果是一樣的。

public class Test {

 // 打印方式跟上一種方式一樣，這里不在給出。
 private int flag = 1;
 private Lock lock = new ReentrantLock();
 private Condition condition1 = lock.newCondition();
 private Condition condition2 = lock.newCondition();
 private Condition condition3 = lock.newCondition();

 private  void  print1() {
  try {
   lock.lock();
   while (flag != 1) {
    try {
     this.condition1.await();
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
   }
   System.out.print("A");
   flag = 2;
   this.condition2.signal();
  }finally {
   lock.unlock();
  }

 }


 private void  print2() {
  try {
   lock.lock();
   while (flag != 2) {
    try {
     this.condition2.await();
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
   }
   System.out.print("B");
   flag = 3;
   this.condition3.signal();
  }finally {
   lock.unlock();
  }

 }

 private void  print3() {
  try {
   lock.lock();
   while (flag != 3) {
    try {
     this.condition3.await();
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
   }
   System.out.print("C");
   flag = 1;
   this.condition1.signal();
  }finally {
   lock.unlock();
  }
 }

Semaphore實(shí)現(xiàn)

信號(hào)量獲取和歸還機(jī)制來保證共享數(shù)據(jù)并發(fā)安全，以下為部分核心代碼；

// 以s1開始的信號(hào)量,初始信號(hào)量數(shù)量為1
private static Semaphore s1 = new Semaphore(1);
// s2、s3信號(hào)量,s1完成后開始,初始信號(hào)數(shù)量為0
private static Semaphore s2 = new Semaphore(0);
private static Semaphore s3 = new Semaphore(0);
static class Thread1 extends Thread {
      @Override
      public void run() {
         try {
            for (int i = 0; i < 10; i++) {
               s1.acquire();// s1獲取信號(hào)執(zhí)行,s1信號(hào)量減1,當(dāng)s1為0時(shí)將無法繼續(xù)獲得該信號(hào)量
               System.out.print("1");
               s2.release();// s2釋放信號(hào)，s2信號(hào)量加1（初始為0），此時(shí)可以獲取B信號(hào)量
             }
         } catch (InterruptedException e) {
            e.printStackTrace();
      }
   }
}

其實(shí)除了以上方法，用CountDownLatch實(shí)現(xiàn)多個(gè)線程互相等待應(yīng)該也是可以解決的，這里不在過多舉例。

Go實(shí)現(xiàn)

在用Go的實(shí)現(xiàn)過程中，主要用到了三個(gè)知識(shí)點(diǎn)。1、先后啟用了三個(gè)goroutine對(duì)共享變量進(jìn)行操作; 2、一把互斥鎖產(chǎn)生的三個(gè)條件變量對(duì)三個(gè)協(xié)程進(jìn)行控制; 3、使用signChannel目的是為了不讓goroutine過早結(jié)束運(yùn)行。

package main

import (
 "log"
 "sync"
)

func main()  {
 //聲明共享變量
 var  flag = 1
 //聲明互斥鎖
 var lock sync.RWMutex
 //三個(gè)條件變量，用于控制三個(gè)協(xié)程執(zhí)行頻率
 cnd1 := sync.NewCond(&lock)
 cnd2 := sync.NewCond(&lock)
 cnd3 := sync.NewCond(&lock)
 //創(chuàng)建一個(gè)通道，用于控制goroutine過早結(jié)束運(yùn)行
 signChannel := make(chan struct{}, 3)
 //最大循環(huán)次數(shù)
 max := 5

 go func(max int) {
  //本次goroutine執(zhí)行完成之后釋放
  defer func() {
   signChannel <- struct{}{}
  }()
  //循環(huán)執(zhí)行
  for i := 1; i <= max; i++ {
   // 鎖定本次臨界環(huán)境變量修改
   lock.Lock()
   //通過for循環(huán)檢測條件是否發(fā)生變化，類似于上面的while
   for flag != 1 {
    //等待
    cnd1.Wait()
   }
   //輸出
   log.Print(flag)
   //修改標(biāo)識(shí)，釋放鎖、并對(duì)其它協(xié)程發(fā)送信號(hào)
   flag = 2
   lock.Unlock()
   cnd2.Signal()
  }
 }(max)

 go func(max int) {
  defer func() {
   signChannel <- struct{}{}
  }()
  for i := 1; i <= max; i++ {
   lock.Lock()
   for flag != 2 {
    cnd2.Wait()
   }
   log.Print(flag)
   flag = 3
   lock.Unlock()
   cnd3.Signal()
  }
 }(max)

 go func(max int) {
  defer func() {
   signChannel <- struct{}{}
  }()
  for i := 1; i <= max; i++ {
   lock.Lock()
   for flag != 3 {
    cnd3.Wait()
   }
   log.Print(flag)
   flag = 1
   lock.Unlock()
   cnd1.Signal()
  }
 }(max)

 <- signChannel
 <- signChannel
 <- signChannel

}

可以看出這種實(shí)現(xiàn)方式也是通過鎖和條件變量來控制臨界區(qū)，這跟線程中Lock、await/signal實(shí)現(xiàn)方式?jīng)]什么區(qū)別。（這是初次學(xué)習(xí)Go中互斥鎖這塊知識(shí)時(shí)，根據(jù)自己理解，編寫的一種實(shí)現(xiàn)方式，如有問題，請多指教或者留言指正）

通過如上加鎖和條件變量的機(jī)制解決了臨界區(qū)變量并發(fā)安全問題，我們知道，之所以會(huì)如上出現(xiàn)并發(fā)問題，從源頭上來說是硬件開發(fā)人員給軟件開發(fā)人員挖的一個(gè)坑，為了提高并發(fā)性能，計(jì)算機(jī)出現(xiàn)了多核CPU，為了提高運(yùn)算速度，CPU中又添加了高速緩存，這就導(dǎo)致多個(gè)CPU在做計(jì)算的時(shí)候緩存不能共享、交替執(zhí)行，從而出現(xiàn)并發(fā)問題，無論線程、還是協(xié)程、解決思路很簡單，通過加鎖、禁用CPU緩存、公用內(nèi)存。當(dāng)然還存在編譯優(yōu)化帶來的指令重排序問題，要想徹底解決必須從編程語言層面保證原子性 、有序性。無論如何處理，要想保證臨界區(qū)變量的安全，總會(huì)存在一定性能損耗。