Go 中的 channel 與 Java BlockingQueue 的本質(zhì)區(qū)別
前言
最近在實(shí)現(xiàn)兩個(gè)需求，由于兩者之間并沒有依賴關(guān)系，所以想利用隊(duì)列進(jìn)行解耦；但在 Go 的標(biāo)準(zhǔn)庫中并沒有現(xiàn)成可用并且并發(fā)安全的數(shù)據(jù)結(jié)構(gòu)；但 Go 提供了一個(gè)更加優(yōu)雅的解決方案，那就是 channel。
channel 應(yīng)用
Go 與 Java 的一個(gè)很大的區(qū)別就是并發(fā)模型不同，Go 采用的是 CSP(Communicating sequential processes) 模型；用 Go 官方的說法：
Do not communicate by sharing memory; instead, share memory by communicating.
翻譯過來就是：不用使用共享內(nèi)存來通信，而是用通信來共享內(nèi)存。
而這里所提到的通信，在 Go 里就是指代的 channel。
只講概念并不能快速的理解與應(yīng)用，所以接下來會(huì)結(jié)合幾個(gè)實(shí)際案例更方便理解。
futrue task
Go 官方?jīng)]有提供類似于 Java 的 FutureTask 支持：
public static void main(String[] args) throws InterruptedException, ExecutionException {
ExecutorService executorService = Executors.newFixedThreadPool(2);
Task task = new Task();
FutureTask<String> futureTask = new FutureTask<>(task);
executorService.submit(futureTask);
String s = futureTask.get();
System.out.println(s);
executorService.shutdown();
}
}
class Task implements Callable<String> {
@Override
public String call() throws Exception {
// 模擬http
System.out.println("http request");
Thread.sleep(1000);
return "request success";
}
}
但我們可以使用 channel 配合 goroutine 實(shí)現(xiàn)類似的功能：
func main() {
ch := Request("https://github.com")
select {
case r := <-ch:
fmt.Println(r)
}
}
func Request(url string) <-chan string {
ch := make(chan string)
go func() {
// 模擬http請(qǐng)求
time.Sleep(time.Second)
ch <- fmt.Sprintf("url=%s, res=%s", url, "ok")
}()
return ch
}
goroutine 發(fā)起請(qǐng)求后直接將這個(gè) channel 返回，調(diào)用方會(huì)在請(qǐng)求響應(yīng)之前一直阻塞，直到 goroutine 拿到了響應(yīng)結(jié)果。
goroutine 互相通信
/**
* 偶數(shù)線程
*/
public static class OuNum implements Runnable {
private TwoThreadWaitNotifySimple number;
public OuNum(TwoThreadWaitNotifySimple number) {
this.number = number;
}
@Override
public void run() {
for (int i = 0; i < 11; i++) {
synchronized (TwoThreadWaitNotifySimple.class) {
if (number.flag) {
if (i % 2 == 0) {
System.out.println(Thread.currentThread().getName() + "+-+偶數(shù)" + i);
number.flag = false;
TwoThreadWaitNotifySimple.class.notify();
}
} else {
try {
TwoThreadWaitNotifySimple.class.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}
/**
* 奇數(shù)線程
*/
public static class JiNum implements Runnable {
private TwoThreadWaitNotifySimple number;
public JiNum(TwoThreadWaitNotifySimple number) {
this.number = number;
}
@Override
public void run() {
for (int i = 0; i < 11; i++) {
synchronized (TwoThreadWaitNotifySimple.class) {
if (!number.flag) {
if (i % 2 == 1) {
System.out.println(Thread.currentThread().getName() + "+-+奇數(shù)" + i);
number.flag = true;
TwoThreadWaitNotifySimple.class.notify();
}
} else {
try {
TwoThreadWaitNotifySimple.class.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}
這里截取了”兩個(gè)線程交替打印奇偶數(shù)“的部分代碼。
Java 提供了 object.wait()/object.notify() 這樣的等待通知機(jī)制，可以實(shí)現(xiàn)兩個(gè)線程間通信。
go 通過 channel 也能實(shí)現(xiàn)相同效果：
func main() {
ch := make(chan struct{})
go func() {
for i := 1; i < 11; i++ {
ch <- struct{}{}
//奇數(shù)
if i%2 == 1 {
fmt.Println("奇數(shù):", i)
}
}
}()
go func() {
for i := 1; i < 11; i++ {
<-ch
if i%2 == 0 {
fmt.Println("偶數(shù):", i)
}
}
}()
time.Sleep(10 * time.Second)
}
本質(zhì)上他們都是利用了線程(goroutine)阻塞然后喚醒的特性，只是 Java 是通過 wait/notify 機(jī)制；
而 go 提供的 channel 也有類似的特性：
向 channel發(fā)送數(shù)據(jù)時(shí)(ch<-struct{}{})會(huì)被阻塞，直到 channel 被消費(fèi)(<-ch)。
以上針對(duì)于
無緩沖 channel。
channel 本身是由 go 原生保證并發(fā)安全的，不用額外的同步措施，可以放心使用。
廣播通知
不僅是兩個(gè) goroutine 之間通信，同樣也能廣播通知，類似于如下 Java 代碼：
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < 10; i++) {
new Thread(() -> {
try {
synchronized (NotifyAll.class){
NotifyAll.class.wait();
}
System.out.println(Thread.currentThread().getName() + "done....");
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
Thread.sleep(3000);
synchronized (NotifyAll.class){
NotifyAll.class.notifyAll();
}
}
主線程將所有等待的子線程全部喚醒，這個(gè)本質(zhì)上也是通過 wait/notify 機(jī)制實(shí)現(xiàn)的，區(qū)別只是通知了所有等待的線程。
換做是 go 的實(shí)現(xiàn)：
func main() {
notify := make(chan struct{})
for i := 0; i < 10; i++ {
go func(i int) {
for {
select {
case <-notify:
fmt.Println("done.......",i)
return
case <-time.After(1 * time.Second):
fmt.Println("wait notify",i)
}
}
}(i)
}
time.Sleep(1 * time.Second)
close(notify)
time.Sleep(3 * time.Second)
}
當(dāng)關(guān)閉一個(gè) channel 后，會(huì)使得所有獲取 channel 的 goroutine 直接返回，不會(huì)阻塞，正是利用這一特性實(shí)現(xiàn)了廣播通知所有 goroutine 的目的。
注意，同一個(gè) channel 不能反復(fù)關(guān)閉，不然會(huì)出現(xiàn)panic。
channel 解耦
以上例子都是基于無緩沖的 channel，通常用于 goroutine 之間的同步；同時(shí) channel 也具備緩沖的特性：
ch :=make(chan T, 100)
可以直接將其理解為隊(duì)列，正是因?yàn)榫哂芯彌_能力，所以我們可以將業(yè)務(wù)之間進(jìn)行解耦，生產(chǎn)方只管往 channel 中丟數(shù)據(jù)，消費(fèi)者只管將數(shù)據(jù)取出后做自己的業(yè)務(wù)。
同時(shí)也具有阻塞隊(duì)列的特性：
當(dāng) channel寫滿時(shí)生產(chǎn)者將會(huì)被阻塞。當(dāng) channel為空時(shí)消費(fèi)者也會(huì)阻塞。
從上文的例子中可以看出，實(shí)現(xiàn)相同的功能 go 的寫法會(huì)更加簡單直接，相對(duì)的 Java 就會(huì)復(fù)雜許多（當(dāng)然這也和這里使用的偏底層 api 有關(guān)）。
Java 中的 BlockingQueue
這些特性都與 Java 中的 BlockingQueue 非常類似，他們具有以下的相同點(diǎn)：
可以通過兩者來進(jìn)行 goroutine/thread通信。具備隊(duì)列的特征，可以解耦業(yè)務(wù)。 支持并發(fā)安全。
同樣的他們又有很大的區(qū)別，從表現(xiàn)上看：
channel支持select語法，對(duì)channel的管理更加簡潔直觀。channel支持關(guān)閉，不能向已關(guān)閉的channel發(fā)送消息。channel支持定義方向，在編譯器的幫助下可以在語義上對(duì)行為的描述更加準(zhǔn)確。
當(dāng)然還有本質(zhì)上的區(qū)別就是 channel 是 go 推薦的 CSP 模型的核心，具有編譯器的支持，可以有很輕量的成本實(shí)現(xiàn)并發(fā)通信。
而 BlockingQueue 對(duì)于 Java 來說只是一個(gè)實(shí)現(xiàn)了并發(fā)安全的數(shù)據(jù)結(jié)構(gòu)，即便不使用它也有其他的通信方式；只是他們都具有阻塞隊(duì)列的特征，所有在初步接觸 channel 時(shí)容易產(chǎn)生混淆。
| 相同點(diǎn) | channel 特有 |
|---|---|
| 阻塞策略 | 支持select |
| 設(shè)置大小 | 支持關(guān)閉 |
| 并發(fā)安全 | 自定義方向 |
| 普通數(shù)據(jù)結(jié)構(gòu) | 編譯器支持 |
總結(jié)
有過一門編程語言的使用經(jīng)歷在學(xué)習(xí)其他語言是確實(shí)是要方便許多，比如之前寫過 Java 再看 Go 時(shí)就會(huì)發(fā)現(xiàn)許多類似之處，只是實(shí)現(xiàn)不同。
拿這里的并發(fā)通信來說，本質(zhì)上是因?yàn)椴l(fā)模型上的不同；
Go 更推薦使用通信來共享內(nèi)存，而 Java 大部分場景都是使用共享內(nèi)存來通信（這樣就得加鎖來同步）。
帶著疑問來學(xué)習(xí)確實(shí)會(huì)事半功倍。
你的點(diǎn)贊與分享是對(duì)我最大的支持