go帶緩沖chan實現(xiàn)消息隊列功能

更新時間：2023年02月28日 10:28:03 作者：wecode.fun

本文主要介紹了go帶緩沖chan實現(xiàn)消息隊列功能，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

1、Channels 定義

通道是一種支持多類型的管道，您可以通過它使用通道運算符 <- 發(fā)送和接收值。

數(shù)據(jù)沿箭頭方向流動。

ch <- v    // Send v to channel ch.
v := <-ch  // Receive from ch, and
           // assign value to v.

與 maps 和 slices 一樣，通道必須在使用前創(chuàng)建：

ch := make(chan int)

默認情況下，發(fā)送和接收阻塞，直到另一方準備就緒。
這允許 goroutines 在沒有顯式鎖或條件變量的情況下進行同步。

package main

import "fmt"

func sum(s []int, c chan int) {
?? ?sum := 0
?? ?for _, v := range s {
?? ??? ?sum += v
?? ?}
?? ?c <- sum // send sum to c
}

func main() {
?? ?s := []int{7, 2, 8, -9, 4, 0}

?? ?c := make(chan int)
?? ?go sum(s[:len(s)/2], c)
?? ?go sum(s[len(s)/2:], c)
?? ?x, y := <-c, <-c // receive from c

?? ?fmt.Println(x, y, x+y)
}

2、chan 常用操作

無緩沖區(qū)：存入讀取一次，存入后未取，再存入就會堵塞，同樣未存，就取也會堵塞。
有緩沖區(qū)：只有當緩沖區(qū)滿了，才會堵塞存；只有緩沖區(qū)空時，才會堵塞取。
len(channel) 返回緩沖區(qū)現(xiàn)有數(shù)據(jù)長度
cap（channel）返回緩沖區(qū)的大小
close(channel) 關(guān)閉 channel，關(guān)閉后，讀取不到數(shù)據(jù)。如下，如果其他協(xié)程關(guān)掉 channel 則會跳出循環(huán)

3、帶緩沖chan實現(xiàn)消息隊列功能

// 監(jiān)測數(shù)據(jù)結(jié)構(gòu)體
type Msg struct {
?? ?Timestamp?? ?int64
?? ?Content?? ??? ?string?? ??? ?
}

// 用 chan 模擬隊列，隊列的元素為 Msg 類型
var SyncQueen chan Msg

// 必須初始化才能使用。初始化一個容量為1024的 chan。chan 滿時會阻塞
func init() {
?? ?SyncQueen = make(chan Msg, 1024)
}

// 隊列消費者
func Consumer() {
?? ?defer func() {
?? ??? ?if err := recover(); err != nil {
?? ??? ??? ?fmt.Println(err)
?? ??? ?}
?? ?}()

?? ?for {
?? ??? ?// chan 內(nèi)無消息則阻塞
?? ??? ?msg := <-SyncQueen

?? ??? ?fmt.Println(msg.Content)
?? ?}
}

// 隊列生產(chǎn)者
func Producer() {
?? ?for {
?? ??? ?msg := Msg(time.now().Unix(), "hello")
?? ??? ?
?? ??? ?// 發(fā)送消息到 chan
?? ??? ?SyncQueen <- msg
?? ??? ?time.Sleep(2 time.Second)
?? ?}
}

重點

多協(xié)程使用chan是并發(fā)安全的，以下展示一個簡單的例子：

// 定義類型為 int 的 chan
var chanNums chan int

// chan 的消費者，用戶后續(xù)多協(xié)程
// 目的：數(shù)組里存儲了10000個數(shù)字，多個協(xié)程并行計算后，把和加起來
func consumer(sum *int) int {
?? ?for {
?? ??? ?v := <-chanNums
?? ??? ?*sum += v
?? ?}
}

//-------------------------------------

func main() {
?? ?var a [10000]int

?? ?for i := 0; i < 10000; i++ {
?? ??? ?a[i] = i + 1
?? ?}

?? ?chanNums = make(chan int, 10000)

?? ?for i := 0; i < 10000; i++ {
?? ??? ?chanNums <- (i + 1)
?? ?}
?? ?var s1, s2, s3, s4, s5 int = 0, 0, 0, 0, 0
?? ?go consumer(&s1)
?? ?go consumer(&s2)
?? ?go consumer(&s3)
?? ?go consumer(&s4)
?? ?go consumer(&s5)

?? ?for {
?? ??? ?time.Sleep(5 * time.Second)
?? ??? ?break
?? ?}

?? ?fmt.Println("s1=", s1, "s2=", s2, "s3=", s3, "s4=", s4, "s5=", s5)
?? ?fmt.Println("sum=", s1+s2+s3+s4+s5)
}

// 輸出
s1= 10818438 s2= 12073966 s3= 9044041 s4= 11509634 s5= 6558921
sum= 50005000

到此這篇關(guān)于go帶緩沖chan實現(xiàn)消息隊列功能的文章就介紹到這了,更多相關(guān)go緩沖chan消息隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: