samber/lo 库的使用方法： 处理 channel

samber/lo 库的使用方法： 处理 channel

samber/lo 是一个 Go 语言库，提供了一些常用的集合操作函数，如 Filter、Map 和 FilterMap。汇总目录页面

这个库函数太多，因此我决定按照功能分别介绍，本文介绍的是 samber/lo 库中channel相关的函数。

ChannelDispatcher

将信息从输入通道消息分发到 N 个子通道中。当输入通道关闭时，这个关闭事件会被传播到所有的子通道，也就是说，所有的子通道也会被关闭。这些通道可以有一个固定的缓冲容量，或者当 cap（容量）为 0 时，它们是无缓冲的。

// 创建了一个带有 42 个缓冲区的整数通道 ch，并向其中发送 0 到 10 的整数。
ch := make(chan int, 42)
for i := 0; i <= 10; i++ {ch <- i
}// 使用 lo.ChannelDispatcher 函数创建了 5 个子通道，每个通道的缓冲区大小为 10。
// 这个函数会将 ch 中的数据按照轮询策略分发到这些子通道中。
children := lo.ChannelDispatcher(ch, 5, 10, DispatchingStrategyRoundRobin[int])
// []<-chan int{...}// 定义了一个 consumer 函数，这个函数会从给定的通道中读取数据，直到通道关闭。
// 如果通道已经关闭，ok 会为 false，我们就打印 "closed" 并退出循环。
consumer := func(c <-chan int) {for {msg, ok := <-cif !ok {println("closed")break}println(msg)}
}// 为每个子通道启动一个 consumer goroutine。
// 这样，我们就创建了 5 个并发的消费者，它们会并行地从 ch 中读取数据。
for i := range children {go consumer(children[i])
}

有很多分发策略可用：

• lo.DispatchingStrategyRoundRobin: 使用轮询策略将消息分发到子通道中。
• lo.DispatchingStrategyRandom: 使用随机策略将消息分发到子通道中。
• lo.DispatchingStrategyWeightedRandom: 使用加权随机策略将消息分发到子通道中。
• lo.DispatchingStrategyFirst: 分发消息到第一个非满的子通道中。
• lo.DispatchingStrategyLeast: 分发消息到最空的子通道中。
• lo.DispatchingStrategyMost: 分发消息到最满的子通道中。

其中一些策略会带有回退机制，以便优先考虑非阻塞行为。请参阅实现。

对于自定义策略，只需实现 lo.DispatchingStrategy 原型即可：

type DispatchingStrategy[T any] func(message T, messageIndex uint64, channels []<-chan T) int

• DispatchingStrategy 是一个函数，它接受三个参数：

• message T：这是要分发的消息，其类型为泛型 T。

• messageIndex uint64：这是消息的索引，通常用于确定将消息分发到哪个通道。

• channels []<-chan T：这是一个通道切片，消息将被分发到这些通道中的一个。

• 这个函数返回一个 int，表示消息应该被分发到 channels 切片中的哪个通道。

Eg:

type Message struct {TenantID uuid.UUID
}func hash(id uuid.UUID) int {h := fnv.New32a()h.Write([]byte(id.String()))return int(h.Sum32())
}// Routes messages per TenantID.
customStrategy := func(message string, messageIndex uint64, channels []<-chan string) int {destination := hash(message) % len(channels)// check if channel is fullif len(channels[destination]) < cap(channels[destination]) {return destination}// fallback when child channel is fullreturn utils.DispatchingStrategyRoundRobin(message, uint64(destination), channels)
}children := lo.ChannelDispatcher(ch, 5, 10, customStrategy)
...

SliceToChannel

返回一个只读的通道，其中包含了集合中的元素。当最后一个元素被读取后，通道会被关闭。第一个参数是通道的容量，第二个参数是集合。

list := []int{1, 2, 3, 4, 5}for v := range lo.SliceToChannel(2, list) {println(v)
}
// prints 1, then 2, then 3, then 4, then 5

ChannelToSlice

返回一个由通道中的元素构建的切片。阻塞直到通道关闭。

list := []int{1, 2, 3, 4, 5}
ch := lo.SliceToChannel(2, list)items := ChannelToSlice(ch)
// []int{1, 2, 3, 4, 5}

Generator

实现了生成器设计模式。通道在最后一个元素被读取后会被关闭。通道的容量可以被定制。 Generator的第一个参数是通道的容量，第二个参数是生成器函数， 返回一个通道。 其中，生成器函数的参数是一个函数，这个函数用于向通道中发送元素。

generator := func(yield func(int)) {yield(1)yield(2)yield(3)
}for v := range lo.Generator(2, generator) {println(v)
}
// prints 1, then 2, then 3

Buffer

创建一个包含 n 个元素的切片，这些元素来自通道。返回切片、切片长度、读取时间和通道状态（打开/关闭）。第一个参数是通道，第二个参数是切片的长度。

ch := lo.SliceToChannel(2, []int{1, 2, 3, 4, 5})items1, length1, duration1, ok1 := lo.Buffer(ch, 3)
// []int{1, 2, 3}, 3, 0s, true
items2, length2, duration2, ok2 := lo.Buffer(ch, 3)
// []int{4, 5}, 2, 0s, false

示例：RabbitMQ 消费者

ch := readFromQueue()for {// read 1k itemsitems, length, _, ok := lo.Buffer(ch, 1000)// do batching stuffif !ok {break}
}

BufferWithTimeout

和Buffer函数类似， 但是增加了一个超时参数， 如果超时，返回已经读取的元素。

generator := func(yield func(int)) {for i := 0; i < 5; i++ {yield(i)time.Sleep(35*time.Millisecond)}
}ch := lo.Generator(0, generator)items1, length1, duration1, ok1 := lo.BufferWithTimeout(ch, 3, 100*time.Millisecond)
// []int{1, 2}, 2, 100ms, true
items2, length2, duration2, ok2 := lo.BufferWithTimeout(ch, 3, 100*time.Millisecond)
// []int{3, 4, 5}, 3, 75ms, true
items3, length3, duration2, ok3 := lo.BufferWithTimeout(ch, 3, 100*time.Millisecond)
// []int{}, 0, 10ms, false

示例：RabbitMQ 消费者

ch := readFromQueue()for {// read 1k items// wait up to 1 seconditems, length, _, ok := lo.BufferWithTimeout(ch, 1000, 1*time.Second)// do batching stuffif !ok {break}
}

示例：多线程的 RabbitMQ 消费者

ch := readFromQueue()// 5 workers
// prefetch 1k messages per worker
children := lo.ChannelDispatcher(ch, 5, 1000, lo.DispatchingStrategyFirst[int])consumer := func(c <-chan int) {for {// read 1k items// wait up to 1 seconditems, length, _, ok := lo.BufferWithTimeout(ch, 1000, 1*time.Second)// do batching stuffif !ok {break}}
}for i := range children {go consumer(children[i])
}

FanIn

合并多个输入通道的消息到一个缓冲通道中。输出消息没有优先级。当所有的上游通道到达 EOF 时，下游通道关闭。

stream1 := make(chan int, 42)
stream2 := make(chan int, 42)
stream3 := make(chan int, 42)all := lo.FanIn(100, stream1, stream2, stream3)
// <-chan int

FanOut

广播所有上游消息到多个下游通道。当上游通道到达 EOF 时，下游通道关闭。如果任何下游通道已满，广播将暂停。

stream := make(chan int, 42)all := lo.FanOut(5, 100, stream)
// [5]<-chan int