image

前言

最近在實(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請求
        time.Sleep(time.Second)
        ch <- fmt.Sprintf("url=%s, res=%s", url, "ok")
    }()
    return ch
}

goroutine 發(fā)起請求后直接將這個(gè) channel 返回乌庶，調(diào)用方會(huì)在請求響應(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 也有類似的特性：

1. 向 channel 發(fā)送數(shù)據(jù)時(shí)(ch<-struct{}{})會(huì)被阻塞，直到 channel 被消費(fèi)(<-ch)酗电。

以上針對于無緩沖 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ì)更加簡單直接拉庵，相對的 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 語法，對 channel 的管理更加簡潔直觀信夫。
• channel 支持關(guān)閉窃蹋，不能向已關(guān)閉的 channel 發(fā)送消息。
• channel 支持定義方向静稻，在編譯器的幫助下可以在語義上對行為的描述更加準(zhǔn)確警没。

當(dāng)然還有本質(zhì)上的區(qū)別就是 channel 是 go 推薦的 CSP 模型的核心，具有編譯器的支持振湾，可以有很輕量的成本實(shí)現(xiàn)并發(fā)通信杀迹。

而 BlockingQueue 對于 Java 來說只是一個(gè)實(shí)現(xiàn)了并發(fā)安全的數(shù)據(jù)結(jié)構(gòu)，即便不使用它也有其他的通信方式押搪；只是他們都具有阻塞隊(duì)列的特征树酪，所有在初步接觸 channel 時(shí)容易產(chǎn)生混淆。

總結(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ì)事半功倍力试。

最近和網(wǎng)友討論后再補(bǔ)充一下，其實(shí) Go channel 的底層實(shí)現(xiàn)也是通過對共享內(nèi)存的加鎖來實(shí)現(xiàn)的排嫌，這點(diǎn)任何語言都不可避免畸裳。

既然都是共享內(nèi)存那和我們自己使用共享內(nèi)存有什么區(qū)別呢？主要還是 channel 的抽象層級更高躏率，我們使用這類高抽象層級的方式編寫代碼會(huì)更易理解和維護(hù)躯畴。

但在一些特殊場景，需要追求極致的性能薇芝，降低加鎖顆粒度時(shí)用共享內(nèi)存會(huì)更加合適蓬抄，所以 Go 官方也提供有sync.Map/Mutex 這樣的庫；只是在并發(fā)場景下更推薦使用 channel 來解決問題夯到。