React单线程处理Channel，多线程异步读取单Channel实现遇阻求助

哈哈，这个问题我太熟了！你遇到的是Raku里react块的一个常见特性——它的事件循环默认是在启动它的那个单一线程里运行的，所以哪怕你用start把整个react包起来，所有whenever的回调也都会挤在同一个线程里依次执行，自然就看不到多线程处理的效果啦。

下面给你两种可行的解决方案，按需选择就行：

方案一：启动多个独立的消费者线程 #

如果你想固定数量的线程同时从Channel读取数据，可以创建多个react块，每个都用start放到独立线程里。Channel会自动把消息分发给处于等待状态的消费者，实现负载均衡：

my Channel $KXGA .= new;
# 往Channel里塞数据
for ^100 { $KXGA.send( (100000..200000).pick ); }
$KXGA.close; # 记得关闭Channel，否则程序最后会挂起等待新消息

# 启动5个消费者线程，你可以根据需求调整数量
for ^5 {
    start react whenever $KXGA -> $number {
        say "In thread ", $*THREAD.id;
        say "→ ", (^$number).sum;
    }
}

# 等待所有任务完成
await Promise.allof($*SCHEDULER.croak-later, $KXGA.closed);

方案二：在回调内部启动异步任务 #

如果希望每次处理数据都用新线程，可以在whenever的回调里嵌套start，把计算逻辑放到异步任务中：

my Channel $KXGA .= new;
for ^100 { $KXGA.send( (100000..200000).pick ); }

my $sums = start react whenever $KXGA -> $number {
    # 把处理逻辑放到独立的异步任务里
    start {
        say "In thread ", $*THREAD.id;
        say "→ ", (^$number).sum;
    }
}

await $sums;

两种方案的区别 #

• 方案一的并发数是固定的，适合控制资源消耗，比如你不想启动太多线程拖垮系统的时候用它；
• 方案二每次处理都会启动新线程（或者复用线程池里的线程），并发量会随消息数量变化，适合对并发数要求不严格的场景，但要注意线程过多带来的开销。

内容的提问来源于stack exchange，提问作者jjmerelo