最新消息:觉得本站不错的话 记得收藏哦 博客内某些功能仅供测试 讨论群:135931704 快养不起小站了 各位有闲钱就打赏下把 My Email weicots#gmail.com Please replace # with @

使用golang的channel做广播

未分类 ajiang-tuzi 183浏览

怎么使用golang的channel做广播
使用golang中的channel做广播需要使用到golang并发模式中的扇出模式,也就是说多个接入点监听一个输入源。这种模式的结果是,只要输入源输入一个消息,任何一个监听者都能获取到这个消息。这里仅有一个例外就是channel关闭。这个关闭将所有监听者都关闭,这就是扇出模式。删除模式简单定义为:多个函数可以从同一个channel读取数据,直到这个channel关闭。

当监听者数量已知时
让每个worker监听专有的广播channel,并且从主channel中派发消息到每一个专有的广播channel中。

type worker struct {
    name   string
    source chan interface{}
    quit   chan struct{}
}

func (w *worker) Start() {
    w.source = make(chan interface{})
    go func() {
        for {
            select {
            case msg := <-w.source:
                fmt.Println("==========>> ", w.name, msg)
            case <-w.quit: // 后面解释
                fmt.Println(w.name, " quit!")
                return
            }
        }
    }()
}

此时定义两个worker:

workers := []*worker{&worker{}, &worker{}}
for _, w := range workers { 
    w.Start() 
}

派发消息:

    go func() {
        msg := "test"
        count := 0
        var sendMsg string
        for {
            select {
            case <-globalQuit:
                fmt.Println("Stop send message")
                return
            case <-time.Tick(500 * time.Millisecond):
                count++
                sendMsg = fmt.Sprintf("%s-%d", msg, count)
                fmt.Println("Send message is ", sendMsg)
                for _, wk := range workers {
                    wk.source <- sendMsg
                }
            }
        }
    }()

当监听者数量为未知时:
在这种情况下,上述解决办法依然可行。唯一不同的地方在于,无论什么时候需要一个新的worker时,仅仅只需要新建一个worker,并开启它,然后push到workers的slice中。但这种方式需要一个线程安全的slice,需要一个同步锁。其实现如下:

type threadSafeSlice struct {
    sync.Mutex
    workers []*worker
}

func (slice *threadSafeSlice) Push(w *worker) {
    slice.Lock()
    defer slice.Unlock()

    slice.workers = append(slice.workers, w)
}

func (slice *threadSafeSlice) Iter(routine func(*worker)) {
    slice.Lock()
    defer slice.Unlock()

    for _, worker := range slice.workers {
        routine(worker)
    }
}

任何时候,需要一个新的worker时:

w := &worker{}
w.Start()
threadSafeSlice.Push(w)

然后派发消息修改如下伪代码所示:

go func() {
    for {
        msg := <- ch
        threadSafeSlice.Iter(func(w *worker) { w.source <- msg })
    }
}()

最后一个问题:绝对不要让一个goroutine挂起
一个好的实践是:绝对不要让一个goroutine挂起。所以当完成监听后,必须关闭所有激活的goroutine。这将通过worker中的quitchannel进行:

首先,创建一个全局的quit信号channel:

globalQuit := make(chan struct{})

并且在任何一个新建的worker时,将globalQuit分配给这个worker的quit channel

worker.quit = globalQuit

然后当需要关闭所有的worker时,仅仅只需要这么做:

close(globalQuit)

因此close将被所有监听的goroutine所接受。所有的goroutine将被返回。

最后完善后的代码如下所示:

package main

import (
    "fmt"
    "sync"
    "time"
)

import (
    "os"
    "os/signal"
    "syscall"
)

type threadSafeSlice struct {
    sync.Mutex
    workers []*worker
}

func (slice *threadSafeSlice) Push(w *worker) {
    slice.Lock()
    defer slice.Unlock()

    slice.workers = append(slice.workers, w)
}

func (slice *threadSafeSlice) Iter(routine func(*worker)) {
    slice.Lock()
    defer slice.Unlock()

    for _, worker := range slice.workers {
        routine(worker)
    }
}

type worker struct {
    name   string
    source chan interface{}
    quit   chan struct{}
}

func (w *worker) Start() {
    w.source = make(chan interface{})
    go func() {
        for {
            select {
            case msg := <-w.source:
                fmt.Println("==========>> ", w.name, msg)
            case <-w.quit: // 后面解释
                fmt.Println(w.name, " quit!")
                return
            }
        }
    }()
}

func main() {
    globalQuit := make(chan struct{})

    tss := &threadSafeSlice{}

    // 1秒钟添加一个新的worker至slice中
    go func() {
        name := "worker"
        for i := 0; i < 10; i++ {
            time.Sleep(1 * time.Second)
            w := &worker{
                name: fmt.Sprintf("%s%d", name, i),
                quit: globalQuit,
            }
            w.Start()
            tss.Push(w)
        }
    }()

    // 派发消息
    go func() {
        msg := "test"
        count := 0
        var sendMsg string
        for {
            select {
            case <-globalQuit:
                fmt.Println("Stop send message")
                return
            case <-time.Tick(500 * time.Millisecond):
                count++
                sendMsg = fmt.Sprintf("%s-%d", msg, count)
                fmt.Println("Send message is ", sendMsg)
                tss.Iter(func(w *worker) { w.source <- sendMsg })
            }
        }
    }()

    // 截获退出信号
    c := make(chan os.Signal, 1)
    signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)

    for sig := range c {
        switch sig {
        case syscall.SIGINT, syscall.SIGTERM: // 获取退出信号时,关闭globalQuit, 让所有监听者退出
            close(globalQuit)
            time.Sleep(1 * time.Second)
            return
        }
    }
}

参考与引用
原文链接怎么使用golang的channel做广播

转载请注明:(●--●) Hello.My Weicot » 使用golang的channel做广播