go带缓冲chan实现消息队列功能_Golang

go带缓冲chan实现消息队列功能

2023-02-28 15:20wecode.fun Golang

本文主要介绍了go带缓冲chan实现消息队列功能，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧

1、Channels 定义

通道是一种支持多类型的管道，您可以通过它使用通道运算符 <- 发送和接收值。

数据沿箭头方向流动。

				?

									ch <- v    // Send v to channel ch.

									v := <-ch  // Receive from ch, and

									           // assign value to v.

与 maps 和 slices 一样，通道必须在使用前创建：

				?

									ch := make(chan int)

默认情况下，发送和接收阻塞，直到另一方准备就绪。
这允许 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 常用操作

无缓冲区：存入读取一次，存入后未取，再存入就会堵塞，同样未存，就取也会堵塞。
有缓冲区：只有当缓冲区满了，才会堵塞存；只有缓冲区空时，才会堵塞取。
len(channel) 返回缓冲区现有数据长度
cap（channel）返回缓冲区的大小
close(channel) 关闭 channel，关闭后，读取不到数据。如下，如果其他协程关掉 channel 则会跳出循环

3、带缓冲chan实现消息队列功能

				?

									// 监测数据结构体

									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 内无消息则阻塞

									        msg := <-SyncQueen

									        fmt.Println(msg.Content)

									    }

									}

				?

									// 队列生产者

									func Producer() {

									    for {

									        msg := Msg(time.now().Unix(), "hello")

									        // 发送消息到 chan

									        SyncQueen <- msg

									        time.Sleep(2 time.Second)

									    }

									}

重点

多协程使用chan是并发安全的，以下展示一个简单的例子：

				?

									// 定义类型为 int 的 chan

									var chanNums chan int

									// chan 的消费者，用户后续多协程

									// 目的：数组里存储了10000个数字，多个协程并行计算后，把和加起来

									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