golang之channel使用简述

pojol · · 622 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

Channel

golang CSP 模型中的C, 主要用于goroutine之间消息的传递,我们知道在写代码的过程中,解偶是非常重要的一环,而使用channel则可以很好的隔离goroutine,使得goroutne之间的交互,只需要将重心关注在如何从channel中消费或者生产消息。
  • 声明和使用
  • 阻塞场景
  • 关闭Channel
  • select & range
  • 使用channel模拟生产消费模型

声明和使用

使用make声明一个channel
    ch := make(chan int)

    ch <- 1    // write    ch位于 <- 的左侧(代表数据流入
    <- ch    // read        ch位于 <- 的右侧(代表数据流出

阻塞场景

在真正使用channel前,我们需要了解channel可能会产生阻塞场景的所有可能,以防止在代码中编写出不符合我们预期的代码。
下面我们罗列出可能的四种情形
无缓冲
channel中无数据,但是执行 <- channel (读
    ch := make(chan interface{})

    <-ch
    fmt.Println("read buf succ")
channel中无数据,往 channel <- (写 ,但是没有goroutine读取。
    ch := make(chan interface{})

    ch <- 1

    fmt.Println("read buf succ")
有缓冲
channel中无数据,但是执行 <- channel
    ch := make(chan interface{}, 1)

    <-ch
    fmt.Println("read buf succ")
channel中已满, 继续执行 channel <- 动作,但是没有goroutine读取。
    ch := make(chan interface{}, 1)

    ch <- 1
    ch <- 2
    fmt.Println("read buf succ")

关闭Channel

使用close关闭channel
    ch := make(chan interface{})
    close(ch)

关闭channel需要注意

  1. 重复关闭会 panic
  2. 向关闭的channel发送数据会panic
  3. 从关闭的channel读取数据,会读取到值的初始值,比如interface类型,读取到的就是nil

select & range

range 字段会阻塞监听 channel, 直到channel 被close。

func recv(ch chan int) {
    for msg := range ch { // 使用 range 可以自动等待 ch 的行为, 直到ch 被close。
        fmt.Println(msg)
    }

    fmt.Println("channel closed")
}

func send(ch chan int, msg int) {
    ch <- msg
}

func main() {
    ch := make(chan int, 2)

    go recv(ch)
    ch <- 1
    ch <- 2
    ch <- 3

    time.AfterFunc(time.Second*2, func() {
        close(ch)
    })
}

select 的大致工作原理

  1. 检查所有的case
  2. 当检查的case已经可以发送|接收,则执行当前代码块
  3. 当有多个case可以执行,则随机选择一个执行
  4. 当没有case可以执行,则阻塞
  5. 如果存在default,当没有可执行代码块时,则执行default代码块

使用select来管理channel的读取, 通过default防止阻塞.

func readCh(ch chan interface{}) error {
    select {
    case v := <-ch:
        fmt.Println(v)
    default:
        return errors.New("no data")
    }

    return nil
}
使用 timer 或者 context 来进行到期退出判定. 另外我们也可以使用sync.Once()这种形式设定一个开关,
来控制select的退出逻辑,可参照grpc/internal/grpcsync/event.go
func readCh(ch chan interface{}) error {
    select {
    case v := <-ch:
        fmt.Println(v)
    case <-time.After(time.Second):
        return errors.New("time arrived")
    }

    return nil
}

使用channel模拟生产消费模型

下面代码的 Unbounded 实现摘自grpc/internal/buffer/unbounded.go,
它没有选择使用带容量的channel,而是另外使用了一个list来备份积压的消息,这里我猜有两个原因

  1. 使用这种方式channel变成了一个任意长度的channel,不用考虑channel被写满导致的问题。
  2. 这里为什么不直接使用list + mutex,因为需要channel的特性来隔离goroutine。

type Unbounded struct {
    c       chan interface{}
    backlog []interface{}

    sync.Mutex
}

func NewUnbounded() *Unbounded {
    return &Unbounded{c: make(chan interface{}, 1)}
}

func (b *Unbounded) Put(t interface{}) {
    b.Lock()
    if len(b.backlog) == 0 {
        select {
        case b.c <- t:
            b.Unlock()
            return
        default:
        }
    }
    b.backlog = append(b.backlog, t)
    b.Unlock()
}

func (b *Unbounded) Load() {
    b.Lock()
    if len(b.backlog) > 0 {
        select {
        case b.c <- b.backlog[0]:
            b.backlog[0] = nil
            b.backlog = b.backlog[1:]
        default:
        }
    }
    b.Unlock()
}

func (b *Unbounded) Get() <-chan interface{} {
    return b.c
}

var q *Queue

type Queue struct {
    buf *Unbounded
}
type QueueInterface interface {
    consume()
    produce(info int)
}

func (q *Queue) consume() {
    for {
        select {
        case t := <-q.buf.Get():
            q.buf.Load()
            fmt.Println(t)
        case <-time.After(time.Second * 10):
            fmt.Println(errors.New("the end"))
        }
    }
}

func (q *Queue) produce(info int) {
    q.buf.Put(info)
}

func main() {

    q := &Queue{
        buf: NewUnbounded(),
    }

    go q.consume()

    q.produce(1)
    q.produce(3)
    time.AfterFunc(time.Second*2, func() {
        for i := 0; i < 3; i++ {
            q.produce(4)
        }
        //q.produce(4)
    })

    select {}
}
注: 这里的实现使用了interface作为channel的消息体,凡是在有性能瓶颈的地方应该使用具体的类型独立实现一版,类似grpc/internal/transport.go中的recvBuffer

有疑问加站长微信联系(非本文作者)

本文来自:Segmentfault

感谢作者:pojol

查看原文:golang之channel使用简述

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

622 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传