如何使用 channel

简介: 如何使用 Channel例子来自于Concurrency is not parallelismGoogle Search: A fake frameworkv1.0var ( Web = fakeSearch("web") Image = fakeSearch("image"...

如何使用 Channel

例子来自于Concurrency is not parallelism

Google Search: A fake framework

v1.0

var (
    Web = fakeSearch("web")
    Image = fakeSearch("image")
    Video = fakeSearch("video")
)

type Search func(query string) Result

func fakeSearch(kind string) Search {
        return func(query string) Result {
              time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
              return Result(fmt.Sprintf("%s result for %q\n", kind, query))
        }
}
func main() {
    rand.Seed(time.Now().UnixNano())
    start := time.Now()
    results := Google("golang")
    elapsed := time.Since(start)
    fmt.Println(results)
    fmt.Println(elapsed)
}

关键函数

func Google(query string) (results []Result) {
    results = append(results, Web(query))
    results = append(results, Image(query))
    results = append(results, Video(query))
    return
}

Google 2.0

每个 search, 独立并发.
No locks. No condition variables. No callbacks.

func Google(query string) (results []Result) {
    c := make(chan Result)
    go func() { c <- Web(query) } ()
    go func() { c <- Image(query) } ()
    go func() { c <- Video(query) } ()

    for i := 0; i < 3; i++ {
        result := <-c
        results = append(results, result)
    }
    return
}

Google 2.1

如果某个服务比较慢,怎么办?
No locks. No condition variables. No callbacks.

func Google(query string) (results []Result) {
    c := make(chan Result)
    go func() { c <- Web(query) } ()
    go func() { c <- Image(query) } ()
    go func() { c <- Video(query) } ()

    timeout := time.After(80 * time.Millisecond)
    for i := 0; i < 3; i++ {
        select {
        case result := <-c:
            results = append(results, result)
        case <-timeout:
            fmt.Println("timed out")
            return
        }
    }
    return
}

Google 3.0 Avoid timeout

No locks. No condition variables. No callbacks.

func First(query string, replicas ...Search) Result {
    c := make(chan Result)
    searchReplica := func(i int) { c <- replicas[i](query) }
    for i := range replicas {
        go searchReplica(i)
    }
    return <-c
}
func Google(query string) (results []Result) {
   c := make(chan Result)
    go func() { c <- First(query, Web1, Web2) } ()
    go func() { c <- First(query, Image1, Image2) } ()
    go func() { c <- First(query, Video1, Video2) } ()
    timeout := time.After(80 * time.Millisecond)
    for i := 0; i < 3; i++ {
        select {
        case result := <-c:
            results = append(results, result)
        case <-timeout:
            fmt.Println("timed out")
            return
        }
    }
    return
}

Google 3.1

上面的例子看起来挺完美,但是存在一个严重的内存泄漏,不知道你看出来没有.
First 中的 searchReplica调用,除了第一个会成功返回以外,其他都不会返回.因为堵塞在 c 上面,从而导致了内存泄漏.
改进也很简单

func First(query string, replicas ...Search) Result {
    c := make(chan Result,len(replicas)) //看似多分配了资源,但是很快就会收回
    searchReplica := func(i int) { c <- replicas[i](query) }
    for i := range replicas {
        go searchReplica(i)
    }
    return <-c
}

经过简单的替换,通过 Go 的并发模型,将一个慢的,顺序执行的,故障敏感的程序改造为了一个快速的,并发的,有冗余的,健壮的程序.

完整的 google 3.1

var (
    Web1 = fakeSearch("web")
    Web2 = fakeSearch("web")
    Image1 = fakeSearch("image")
    Image2 = fakeSearch("image")
    Video1 = fakeSearch("video")
    Video2 = fakeSearch("video")
)

type Search func(query string) Result

func fakeSearch(kind string) Search {
        return func(query string) Result {
              time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
              return Result(fmt.Sprintf("%s result for %q\n", kind, query))
        }
}
func main() {
    rand.Seed(time.Now().UnixNano())
    start := time.Now()
    results := Google("golang")
    elapsed := time.Since(start)
    fmt.Println(results)
    fmt.Println(elapsed)
}

func First(query string, replicas ...Search) Result {
    c := make(chan Result,len(replicas)) 
    searchReplica := func(i int) { c <- replicas[i](query) }
    for i := range replicas {
        go searchReplica(i)
    }
    return <-c
}
func Google(query string) (results []Result) {
   c := make(chan Result)
    go func() { c <- First(query, Web1, Web2) } ()
    go func() { c <- First(query, Image1, Image2) } ()
    go func() { c <- First(query, Video1, Video2) } ()
    timeout := time.After(80 * time.Millisecond)
    for i := 0; i < 3; i++ {
        select {
        case result := <-c:
            results = append(results, result)
        case <-timeout:
            fmt.Println("timed out")
            return
        }
    }
    return
}
目录
相关文章
|
缓存 中间件 流计算
如何解决 Netty Channel.isWritable 返回 false
在 Netty 里,有4个方法用来查询 Channel 的状态:isOpen,isRegistered,isActive,isWritable,其中,isWritable 在并发量很高时会返回很多 false。 isWritable 是什么含义? isWritable:Returns true if and only if the I/O thread will perform the req
2789 0
如何解决 Netty Channel.isWritable 返回 false
|
3月前
|
消息中间件 Kafka Go
从Go channel中批量读取数据
从Go channel中批量读取数据
|
5月前
|
调度 PHP
Swoole 源码分析之 Channel 通道模块
通道,用于协程间通讯,支持多生产者协程和多消费者协程。底层自动实现了协程的切换和调度。通道与 PHP 的 Array 类似,仅占用内存,没有其他额外的资源申请,所有操作均为内存操作,无 IO 消耗。
111 5
Swoole 源码分析之 Channel 通道模块
|
3月前
|
Go
在go中监听多个channel
在go中监听多个channel
|
5月前
|
存储 Go
Go 语言当中 CHANNEL 缓冲
Go 语言当中 CHANNEL 缓冲
|
6月前
|
传感器 JSON Dart
Dart笔记:stream_channel 包用法
Dart笔记:stream_channel 包用法
167 0
|
6月前
|
API
Broadcast Channel API 使用方式
使用Broadcast Channel API,页面A创建并发送消息到&#39;myChannel&#39;,而页面B在同一域名下监听该通道,接收并打印消息。当不再使用时,应关闭通道以释放资源。此示例展示了API的基本用法,可按需扩展。
|
11月前
|
Cloud Native Java Go
这些 channel 用法你都用起来了吗?
这些 channel 用法你都用起来了吗?
|
消息中间件 数据库
Rabbmit channel.QueueDeclare参数初识
接触MQ易经有一段时间了,对QueueDeclare一直没有一个全面的认识。
210 0
|
网络协议 前端开发 UED
Netty之服务端channel的初始化
Netty之服务端channel的初始化
115 0