前言
最近在實現兩個需求,由於兩者之間並沒有依賴關系,所以想利用隊列進行解耦;但在 Go
的標准庫中並沒有現成可用並且並發安全的數據結構;但 Go
提供了一個更加優雅的解決方案,那就是 channel
。
channel 應用
Go
與 Java
的一個很大的區別就是並發模型不同,Go 采用的是 CSP(Communicating sequential processes)
模型;用 Go 官方的說法:
Do not communicate by sharing memory; instead, share memory by communicating.
翻譯過來就是:不用使用共享內存來通信,而是用通信來共享內存。
而這里所提到的通信,在 Go 里就是指代的 channel
。
只講概念並不能快速的理解與應用,所以接下來會結合幾個實際案例更方便理解。
futrue task
Go
官方沒有提供類似於 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
實現類似的功能:
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
發起請求后直接將這個 channel
返回,調用方會在請求響應之前一直阻塞,直到 goroutine
拿到了響應結果。
goroutine 互相通信
/**
* 偶數線程
*/
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() + "+-+偶數" + i);
number.flag = false;
TwoThreadWaitNotifySimple.class.notify();
}
} else {
try {
TwoThreadWaitNotifySimple.class.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}
/**
* 奇數線程
*/
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() + "+-+奇數" + i);
number.flag = true;
TwoThreadWaitNotifySimple.class.notify();
}
} else {
try {
TwoThreadWaitNotifySimple.class.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}
這里截取了”兩個線程交替打印奇偶數“的部分代碼。
Java 提供了 object.wait()/object.notify()
這樣的等待通知機制,可以實現兩個線程間通信。
go
通過 channel
也能實現相同效果:
func main() {
ch := make(chan struct{})
go func() {
for i := 1; i < 11; i++ {
ch <- struct{}{}
//奇數
if i%2 == 1 {
fmt.Println("奇數:", i)
}
}
}()
go func() {
for i := 1; i < 11; i++ {
<-ch
if i%2 == 0 {
fmt.Println("偶數:", i)
}
}
}()
time.Sleep(10 * time.Second)
}
本質上他們都是利用了線程(goroutine
)阻塞然后喚醒的特性,只是 Java 是通過 wait/notify 機制;
而 go 提供的 channel 也有類似的特性:
- 向
channel
發送數據時(ch<-struct{}{}
)會被阻塞,直到 channel 被消費(<-ch
)。
以上針對於
無緩沖 channel
。
channel
本身是由 go
原生保證並發安全的,不用額外的同步措施,可以放心使用。
廣播通知
不僅是兩個 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();
}
}
主線程將所有等待的子線程全部喚醒,這個本質上也是通過 wait/notify
機制實現的,區別只是通知了所有等待的線程。
換做是 go
的實現:
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)
}
當關閉一個 channel
后,會使得所有獲取 channel
的 goroutine
直接返回,不會阻塞,正是利用這一特性實現了廣播通知所有 goroutine
的目的。
注意,同一個 channel 不能反復關閉,不然會出現panic。
channel 解耦
以上例子都是基於無緩沖的 channel
,通常用於 goroutine
之間的同步;同時 channel 也具備緩沖的特性:
ch :=make(chan T, 100)
可以直接將其理解為隊列,正是因為具有緩沖能力,所以我們可以將業務之間進行解耦,生產方只管往 channel
中丟數據,消費者只管將數據取出后做自己的業務。
同時也具有阻塞隊列的特性:
- 當
channel
寫滿時生產者將會被阻塞。 - 當
channel
為空時消費者也會阻塞。
從上文的例子中可以看出,實現相同的功能 go 的寫法會更加簡單直接,相對的 Java 就會復雜許多(當然這也和這里使用的偏底層 api 有關)。
Java 中的 BlockingQueue
這些特性都與 Java 中的 BlockingQueue
非常類似,他們具有以下的相同點:
- 可以通過兩者來進行
goroutine/thread
通信。 - 具備隊列的特征,可以解耦業務。
- 支持並發安全。
同樣的他們又有很大的區別,從表現上看:
channel
支持select
語法,對channel
的管理更加簡潔直觀。channel
支持關閉,不能向已關閉的channel
發送消息。channel
支持定義方向,在編譯器的幫助下可以在語義上對行為的描述更加准確。
當然還有本質上的區別就是 channel 是 go 推薦的 CSP
模型的核心,具有編譯器的支持,可以有很輕量的成本實現並發通信。
而 BlockingQueue
對於 Java
來說只是一個實現了並發安全的數據結構,即便不使用它也有其他的通信方式;只是他們都具有阻塞隊列的特征,所有在初步接觸 channel
時容易產生混淆。
相同點 | channel 特有 |
---|---|
阻塞策略 | 支持select |
設置大小 | 支持關閉 |
並發安全 | 自定義方向 |
普通數據結構 | 編譯器支持 |
總結
有過一門編程語言的使用經歷在學習其他語言是確實是要方便許多,比如之前寫過 Java
再看 Go
時就會發現許多類似之處,只是實現不同。
拿這里的並發通信來說,本質上是因為並發模型上的不同;
Go
更推薦使用通信來共享內存,而 Java
大部分場景都是使用共享內存來通信(這樣就得加鎖來同步)。
帶着疑問來學習確實會事半功倍。
最近和網友討論后再補充一下,其實 Go channel
的底層實現也是通過對共享內存的加鎖來實現的,這點任何語言都不可避免。
既然都是共享內存那和我們自己使用共享內存有什么區別呢?主要還是 channel 的抽象層級更高,我們使用這類高抽象層級的方式編寫代碼會更易理解和維護。
但在一些特殊場景,需要追求極致的性能,降低加鎖顆粒度時用共享內存會更加合適,所以 Go 官方也提供有 sync.Map/Mutex
這樣的庫;只是在並發場景下更推薦使用 channel
來解決問題。