开发工具【免费下载链接】concurrent-rubyModern concurrency tools including agents, futures, promises, thread pools, supervisors, and more. Inspired by Erlang, Clojure, Scala, Go, Java, JavaScript, and classic concurrency patterns.项目地址https://gitcode.com/gh_mirrors/co/concurrent-ruby点击查看免费下载本篇指南以 concurrent-ruby 仓库中的官方通道文档docs-source/channel.out.md该文件被 通道实现源码 以{include:file:...}方式嵌入为类文档注释为骨架完整讲解 Edge 版Concurrent::Promises::Channel的创建、阻塞式 push/pop、_op后缀的 Promise 集成、多通道选择select、非阻塞try_变体、超时选项、背压系统以及零容量通道的工作线程同步。读完本文你将掌握用有界通道构建生产者自动减速的自调节系统并用 Future 链取代线程实现免线程分配的生产者/消费者流水线。适用前提Concurrent::Promises::Channel属于concurrent-ruby-edge实验性扩展包需通过require concurrent-edge加载示例基于 channel.init.rb 中的初始化环境其中do_stuff被定义为sleep 0.01以模拟消费者耗时。一、创建通道容量与默认值先创建一个容量为 2 的有界通道ch Concurrent::Promises::Channel.new 2 # #Concurrent::Promises::Channel:0x000002 capacity taken 0 of 2构造参数capacity是可选的整数表示通道中最多可暂存的消息数不传时默认值为UNLIMITED_CAPACITY见 源码第 35-46 行即无限容量此时 push 永不因满而阻塞。to_s/inspect输出capacity taken X of Y其中X为当前size已存消息数Y为容量容量无限时显示unlimited。测试用例在 spec/concurrent/edge/channel_spec.rb 中验证了这一输出格式。内部实现使用一个Mutex保护三个核心数据结构Probes等待中的 pop/select 探针、Messages缓冲区消息队列、PendingPush因满而挂起的 push 操作见 初始化方法。从源码结构看当前实现是加锁的互斥版本源码注释也记录了未来替换为无锁实现的计划第 23 行 TODO。二、阻塞式 push/pop满时挂起、取走即恢复通道是严格的 FIFO先进先出队列push家族方法送入消息pop家族方法取出消息。向容量为 2 的通道推入 3 条消息可以观察到第 3 个线程因通道已满而进入睡眠threads Array.new(3) { |i| Thread.new { ch.push message: i } } sleep 0.01 # let the threads run threads # [#Thread:0x000003channel.in.md:14 dead, # #Thread:0x000004channel.in.md:14 dead, # #Thread:0x000005channel.in.md:14 sleep_forever]当消息被弹出后最后一个线程得以继续并正常结束ch.pop # {:message1} threads.map(:join) # [#Thread:0x000003channel.in.md:14 dead, # #Thread:0x000004channel.in.md:14 dead, # #Thread:0x000005channel.in.md:14 dead]取出方向的阻塞同理。此时通道中剩 2 条消息创建 3 个线程尝试 pop其中一个会一直阻塞直到有新消息推入threads Array.new(3) { |i| Thread.new { ch.pop } } sleep 0.01 # let the threads run threads # [#Thread:0x000006channel.in.md:32 dead, # #Thread:0x000007channel.in.md:32 dead, # #Thread:0x000008channel.in.md:32 sleep_forever] ch.push message: 3 # #Concurrent::Promises::Channel:0x000002 capacity taken 0 of 2 threads.map(:value) # [{:message0}, {:message2}, {:message3}]底层原理push会先调用内部方法ns_try_push——它优先尝试与Probes中等待的 pop 探针配对探针存在则直接交付消息并返回true配对失败时若Capacity Messages.size就把消息推入Messages否则返回false表示满见 ns_try_push 实现。当ns_try_push返回false时阻塞式push会创建一个resolvable_future挂入PendingPush然后在wait!上睡眠直到对应的 pop 出现push 实现。注意示例中push message: i实际推送的是一个 Hash 键值对{message: i}因此pop返回{:message1}而push 1推送的是数字。三、Promises 集成push_op / pop_op 把操作变成 Future该通道的核心特性是与 Promises 深度集成所有操作都可以表示为 Future由push_op/pop_op返回。同样向容量 2 的通道发起 3 次 push 操作前两个立即完成第三个保持 pendingch Concurrent::Promises::Channel.new 2 # #Concurrent::Promises::Channel:0x000009 capacity taken 0 of 2 push_operations Array.new(3) { |i| ch.push_op message: i } # [#Concurrent::Promises::Future:0x00000a fulfilled with #Concurrent::Promises::Channel:0x000009 capacity taken 2 of 2, # #Concurrent::Promises::Future:0x00000b fulfilled with #Concurrent::Promises::Channel:0x000009 capacity taken 2 of 2, # #Concurrent::Promises::ResolvableFuture:0x00000c pending]这里不需要像 Thread 版本那样sleep等待执行容量恰好能容纳 2 条消息因此两个 Promises立即解析全程未分配任何线程。push/pop 操作通常比线程方式更高效。剩余那个 pending 的 push 操作也永远不会再需要线程——当消息被 pop、通道腾出空间时它会被自动解析。ch.pop_op.value! # {:message0} push_operations.map(:value!) # [#Concurrent::Promises::Channel:0x000009 capacity taken 2 of 2, # #Concurrent::Promises::Channel:0x000009 capacity taken 2 of 2, # #Concurrent::Promises::Channel:0x000009 capacity taken 2 of 2] pop_operations Array.new(3) { |i| ch.pop_op } # [#Concurrent::Promises::ResolvableFuture:0x00000d fulfilled with {:message1}, # #Concurrent::Promises::ResolvableFuture:0x00000e fulfilled with {:message2}, # #Concurrent::Promises::ResolvableFuture:0x00000f pending] ch.push message: 3 # (push|pop) 可与 (push_op|pop_op) 自由混用 pop_operations.map(:value) # [{:message1}, {:message2}, {:message3}]源码级验证push_op(message)先尝试ns_try_push成功则返回Promises.fulfilled_future self失败则创建resolvable_future存入PendingPush后返回push_op 实现。pop_op(probe Promises.resolvable_future)委托给内部方法ns_pop_op有缓冲消息则立即 fulfill没有则尝试与PendingPush中挂起的 push 原子配对同时 fulfill 双方见 ns_pop_op 实现。Future 还提供快捷方法then_channel_push(channel)将本 Future 的值推入目标通道并返回推送完成的扁平化 Future定义于 edge/promises.rb 的 NewChannelIntegration 模块self.then(channel) { |value, ch| ch.push_op value }.flat_future。源码文档特别提醒对pop_op用wait(1)超时不会阻止通道稍后完成该操作。要么后续继续处理如pop_op.then { |message| ... }要么用channel.pop_op.wait(1, [true, nil, nil])在超时时主动 fulfill 该操作、防止通道执行弹出见 源码注释。四、多通道选择select / select_op当需要同时监听多个通道时可用类工厂方法.select实例方法等价写法为ch1.select(ch2)。它会以首个有消息可取的通道完成并返回[channel, message]二元组以便区分消息来自哪个通道ch1 Concurrent::Promises::Channel.new 2 # #Concurrent::Promises::Channel:0x000010 capacity taken 0 of 2 ch2 Concurrent::Promises::Channel.new 2 # #Concurrent::Promises::Channel:0x000011 capacity taken 0 of 2 ch1.push 1 # #Concurrent::Promises::Channel:0x000010 capacity taken 1 of 2 ch2.push 2 # #Concurrent::Promises::Channel:0x000011 capacity taken 1 of 2 Concurrent::Promises::Channel.select([ch1, ch2]) # [#Concurrent::Promises::Channel:0x000010 capacity taken 0 of 2, 1] ch1.select(ch2) # [#Concurrent::Promises::Channel:0x000011 capacity taken 0 of 2, 2]配合 Future 链可以写出优雅的多路复用Concurrent::Promises.future { 3 4 }.then_channel_push(ch1) # #Concurrent::Promises::Future:0x000012 pending Concurrent::Promises::Channel. # 或等价写法 ch1.select_op(ch2) select_op([ch1, ch2]). then(got number %03d from ch%d) { |(channel, value), format| format format, value, [ch1, ch2].index(channel).succ }.value! # got number 007 from ch1说明与实现细节文档早期版本将入口称为.select_channel工厂方法当前 API 中实际对应的是类方法.select、.select_op、.try_select定义于 类方法区以及实例方法select、select_op、try_select所有*_matching变体还支持按#匹配器过滤消息。类方法select_op(channels)实现为channels.first.select_op(channels[1..-1], probe)即把第一个通道作为接收者、其余作为候选通道列表实例方法select_op_matching会对[self, *channels]逐个调用partial_select_op注册同一个 probeselect_op_matching 实现任一通道先有消息则整个 probe 完成从而保证只消费一条、只命中一个通道。上述例子中then_channel_push(ch1)先由future { 3 4 }计算出 7经select_op从 ch1 取到[ch1, 7]再由then块格式化输出got number 007 from ch1完整展示了计算 → 推送 → 多路选择 → 格式化的 Future 流水线。五、非阻塞变体try_push / try_pop / try_select所有阻塞操作pop、push、select都有以try_前缀的非阻塞版本立即返回用返回值直接指示成功或失败绝不挂起。ch # #Concurrent::Promises::Channel:0x000009 capacity taken 0 of 2 ch.try_push 1 # true ch.try_push 2 # true ch.try_push 3 # false ch.try_pop # 1 ch.try_pop # 2 ch.try_pop # niltry_push(message)在互斥区内调用ns_try_push返回true推入成功/false通道已满见 try_push。try_pop(no_value nil)内部调用try_pop_matching(ANY, no_value)通道为空时返回no_value默认nil它还顺带尝试消费PendingPush中已挂起的 push见 try_pop_matching。同理try_select(channels)遍历[self, *channels]依次try_pop_matching任一通道有消息即返回[channel, message]全部为空则返回niltry_select_matching。测试中对双通道皆空返回 nil、单通道有值返回该通道消息有完整断言spec 第 186-191 行。另外还有只读不取的peek/peek_matching行为同try_pop但不移除消息见 peek。六、超时选项push / pop / select 的时限语义所有阻塞操作都接受可选的timeout秒Numeric参数。使用超时后返回值本身就能区分成功与超时ch # #Concurrent::Promises::Channel:0x000009 capacity taken 0 of 2 ch.push 1, 0.01 # true ch.push 2, 0.01 # true ch.push 3, 0.01 # false ch.pop 0.01 # 1 ch.pop 0.01 # 2 ch.pop 0.01 # nil返回值语义与 源码注释 一致push(message, timeout)未指定 timeout 时成功返回self通道对象指定 timeout 时成功返回true、超时返回false。pop(timeout nil, timeout_value nil)返回消息本身超时返回timeout_value默认nil。也支持自定义超时值例如pop 0.01, :timed_out。select(channels, timeout nil)返回[channel, message]超时返回nil。阻塞实现最终都收敛为对底层 Future 的wait!/value!带超时调用见 push 与 pop测试对push 超时返回 false、pop 超时返回 nil、不超时正常返回均有覆盖spec 第 43-55、95-109 行。七、背压系统线程版生产者自动减速通道最重要的应用场景是构建带背压backpressure的系统一个自我调节的系统当消费者跟不上时生产者会自然放慢。channel Concurrent::Promises::Channel.new 2 # #Concurrent::Promises::Channel:0x000013 capacity taken 0 of 2 log Concurrent::Array.new # [] producers Array.new 2 do |i| Thread.new(i) do |i| 4.times do |j| log.push format producer %d pushing %d, i, j channel.push [i, j] end end end # [#Thread:0x000014channel.in.md:133 run, # #Thread:0x000015channel.in.md:133 run] consumers Array.new 4 do |i| Thread.new(i) do |consumer| 2.times do |j| from, message channel.pop log.push format consumer %d got %d. payload %d from producer %d, consumer, j, message, from do_stuff end end end # [#Thread:0x000016channel.in.md:142 run, # #Thread:0x000017channel.in.md:142 run, # #Thread:0x000018channel.in.md:142 run, # #Thread:0x000019channel.in.md:142 run] # wait for all to finish producers.map(:join) # [#Thread:0x000014channel.in.md:133 dead, # #Thread:0x000015channel.in.md:133 dead] consumers.map(:join) # [#Thread:0x000016channel.in.md:142 dead, # #Thread:0x000017channel.in.md:142 dead, # #Thread:0x000018channel.in.md:142 dead, # #Thread:0x000019channel.in.md:142 dead] # investigate log log # [producer 0 pushing 0, # producer 0 pushing 1, # producer 0 pushing 2, # producer 1 pushing 0, # consumer 0 got 0. payload 0 from producer 0, # producer 0 pushing 3, # consumer 1 got 0. payload 1 from producer 0, # consumer 2 got 0. payload 2 from producer 0, # consumer 3 got 0. payload 0 from producer 1, # producer 1 pushing 1, # producer 1 pushing 2, # consumer 1 got 1. payload 3 from producer 0, # producer 1 pushing 3, # consumer 2 got 1. payload 1 from producer 1, # consumer 3 got 1. payload 2 from producer 1, # consumer 0 got 1. payload 3 from producer 1]观察日志可以发现生产者远快于消费者消费者每次都要执行do_stuff耗时操作但通道填满后生产者便不再继续 push而是阻塞减速直到消费者腾出空间——消息总量始终不会超过通道容量这里为 2这就是背压的直观表现。初始环境脚本在 channel.init.rb 中把do_stuff定义为sleep 0.01正是为了放大消费者慢的效果。八、背压系统Promise 版用线程池替代常驻线程如果不想为每个生产者和消费者永久分配线程可以用 Promise 取代线程让它们运行在线程池上channel Concurrent::Promises::Channel.new 2 # #Concurrent::Promises::Channel:0x00001a capacity taken 0 of 2 log Concurrent::Array.new # [] def produce(channel, log, producer, i) log.push format producer %d pushing %d, producer, i channel.push_op([producer, i]).then do i 1 4 ? produce(channel, log, producer, i 1) : :done end end # :produce def consume(channel, log, consumer, i) channel.pop_op.then(consumer, i) do |(from, message), consumer, i| log.push format consumer %d got %d. payload %d from producer %d, consumer, i, message, from do_stuff i 1 2 ? consume(channel, log, consumer, i 1) : :done end end # :consume producers Array.new 2 do |i| Concurrent::Promises.future(channel, log, i) { |*args| produce *args, 0 }.run end # [#Concurrent::Promises::Future:0x00001b pending, # #Concurrent::Promises::Future:0x00001c pending] consumers Array.new 4 do |i| Concurrent::Promises.future(channel, log, i) { |*args| consume *args, 0 }.run end # [#Concurrent::Promises::Future:0x00001d pending, # #Concurrent::Promises::Future:0x00001e pending, # #Concurrent::Promises::Future:0x00001f pending, # #Concurrent::Promises::Future:0x000020 pending] # wait for all to finish producers.map(:value!) # [:done, :done] consumers.map(:value!) # [:done, :done, :done, :done] # investigate log log # [producer 0 pushing 0, # producer 1 pushing 0, # producer 1 pushing 1, # consumer 1 got 0. payload 0 from producer 1, # consumer 2 got 0. payload 1 from producer 1, # producer 0 pushing 1, # producer 0 pushing 2, # producer 0 pushing 3, # producer 1 pushing 2, # consumer 0 got 0. payload 0 from producer 0, # consumer 3 got 0. payload 1 from producer 0, # producer 1 pushing 3, # consumer 2 got 1. payload 2 from producer 0, # consumer 1 got 1. payload 3 from producer 0, # consumer 3 got 1. payload 3 from producer 1, # consumer 0 got 1. payload 2 from producer 1]这段代码展示了一种递归的 Future 链设计produce每次push_op完成后用.then触发下一次推送直到推满 4 条返回:doneconsume每次pop_op拿到[from, message]后执行do_stuff再递归发起下一次pop_op2 个生产者 4 个消费者以 Future 形式提交到默认线程池Concurrent::Promises.future{...}.run全程没有为每个角色分配专用线程日志同样显示生产者会因通道满而暂缓push_op保持 pending由后续的 pop 自动配对解析——背压机制与线程版完全一致但资源占用更省。该模式在测试中被进一步验证为push 与 pop 的任意混用组合都能正确配对exchanging 测试。九、零容量通道通过传递值实现工作线程同步若容量为0则任何 push 操作只有在存在可与之配对的 pop 操作时才能成功——操作必须成对才能完成这正是无缓冲通道类似 Go 中不带容量参数的 channel的语义可用于工作线程之间的值传递同步channel Concurrent::Promises::Channel.new 0 # #Concurrent::Promises::Channel:0x000021 capacity taken 0 of 0 thread Thread.new { channel.pop }; sleep 0.01 # allow the thread to go to sleep thread # #Thread:0x000022channel.in.md:214 sleep_forever # succeeds because there is matching pop operation waiting in the thread channel.try_push(:v1) # true # remains pending, since there is no matching operation push channel.push_op(:v2) # #Concurrent::Promises::ResolvableFuture:0x000023 pending thread.value # :v1 # the push operation resolves as a pairing pop is called channel.pop # :v2 push # #Concurrent::Promises::ResolvableFuture:0x000023 fulfilled with #Concurrent::Promises::Channel:0x000021 capacity taken 0 of 0原理容量为 0 时Capacity Messages.size恒为假ns_try_push无法把消息存入缓冲只能尝试与等待的 pop 探针配对。因此try_push(:v1)成功是因为已有一个pop线程在Probes中等待消息被直接交付给该线程thread.value返回:v1push_op(:v2)没有可配对的 pop于是作为ResolvableFuture挂入PendingPush保持 pending随后的channel.pop与挂起的 push 原子配对pushFuture 随即 fulfilled。测试中的push_first/pop_first辅助方法系统性地覆盖了先 push 后 pop与先 pop 后 push共 12 种 push/pop 方法组合spec 第 245-346 行验证零容量通道下所有操作形态都能正确成对交换。十、补充匹配器变体与两个 Channel 的边界匹配器matcher变体除ANY匹配一切外pop家族还提供pop_matching(matcher, ...)、try_pop_matching(matcher, ...)、pop_op_matching(matcher, ...)通过matcher message只取出符合条件的消息matcher可为Symbol、正则、类等任何实现#的对象。测试在通道中混入junk后用pop_matching(Symbol)精确取出:v1spec 第 123-180 行。两个 Channel 的边界本仓库还维护着一个更早的、仿 Go 风格的Concurrent::Channel实现于 lib/concurrent-ruby-edge/concurrent/channel.rb提供put/take//~、Channel.go、四种缓冲类型unbuffered/buffered/dropping/sliding、select块、close/each、timer/ticker等其文档为 docs-source/channel.md配套的 Go 移植示例见 examples/a-tour-of-go-channels 与 examples/go-by-example-channels。而本文讲解的Concurrent::Promises::Channel则是与 Promises/Future 深度集成的演进版本两者并存于 edge 命名空间均通过 concurrent-edge.rb 加载本指南聚焦后者。总结与延伸阅读Concurrent::Promises::Channel的价值在于把队列通信与Promise 链统一阻塞操作push/pop/select提供传统的线程间同步_op后缀方法把每个操作变成可组合的 Future天然支持背压与流水线且大量操作无需分配线程即可完成。从 edge/channel.rb 的源码注释可以看出关闭closing通道等能力仍在 TODO 规划中当前实现也是互斥锁版本——使用时需注意其 edge 属性与相应的限制。核心实现lib/concurrent-ruby-edge/concurrent/edge/channel.rbPromise 集成快捷方法then_channel_pushlib/concurrent-ruby-edge/concurrent/edge/promises.rb完整测试用例spec/concurrent/edge/channel_spec.rb官方文档输入/初始化脚本docs-source/channel.in.md、docs-source/channel.init.rb经典 Go 风格通道对照阅读docs-source/channel.md 与 lib/concurrent-ruby-edge/concurrent/channel.rb赞分享开发工具【免费下载链接】concurrent-rubyModern concurrency tools including agents, futures, promises, thread pools, supervisors, and more. Inspired by Erlang, Clojure, Scala, Go, Java, JavaScript, and classic concurrency patterns.项目地址https://gitcode.com/gh_mirrors/co/concurrent-ruby点击查看免费下载相关推荐concurrent-ruby 通道Channel实战指南Promises 集成的阻塞/非阻塞通信、背压与多通道选择concurrent ruby 通道Channel实战指南Promises 集成的阻塞/非阻塞通信、背压与多通道选择 导读 本文围绕 concurrent开发工具如何安装和配置vulscan完整NSE脚本部署教程如何安装和配置vulscan完整NSE脚本部署教程 vulscan是一款强大的Nmap NSENmap脚本引擎模块它能将Nmap提升为专业的漏洞扫描工具网络安全应用安全Concurrent Ruby 并发工具库完全指南线程安全抽象、线程池、Promises 与安装实战Concurrent Ruby 并发工具库完全指南线程安全抽象、线程池、Promises 与安装实战 导读 Concurrent Ruby https://l开发工具上一篇DBeaver SQL编辑器标签标题自定义配置指南下一篇解决TranslucentTB任务栏透明工具Hook设置失败的终极方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考