Ruby并发-Queue and SizedQueue

Posted 2021-04-28 ruby程序员

Queue and SizedQueue

开门见山，首先来看一段代码：

list.each do |item|

do_something(item)

end

So，你觉得你段代码有什么问题嘛？我来举个例子，假设每次迭代都很耗时，有可能是耗时的计算，有可能是http请求

list.each do |item|

do_something_slow(item)

end

怎么办？很明显，使用多线程。对于每一次迭代我们都启用一个线程来执行，最后我们把他们join在一起，也就是说等待他们全部完成。

list.map do |item|

Thread.new do

do_something(item)

end

end.each(&:join)

现在又有另外一个问题，如果这个list很大呢？有可能是一个大的文件，有可能是很大的stream

huge_list = File.open('huge.txt')

huge_list = TCPSocket.new('somewhere.com', 1234)

如果还用上面的方法，会生成非常多的线程，那么最简单的方式是使用线程池，在Ruby的标准类库种已经为我们提供了这样的gems

Meet Queue

Queue是一个线程安全的数组，无论哪个线程在读或者在写，状态总是保持一致的，这里我们设置一个容量为20的线程池，这20个线程从Queue中获取任务并执行，我们把所有的item放入Queue中，线程就会一个一个的去Queue中取，直到所有任务完成。

Great！但是这段代码还有个问题，生产者queue和消费者thread独立运行，但是如果生产者的生产速度快于消费者的消费速度，items会在queue中积累，但是我们又不想生成过多的thread来消耗内存。SizedQueue上场！

Meet SizedQueue.

这个例子与上面的类似，唯一不同的是使用SizedQueue代替Queue，SizedQueue可以设定容量，当容量满了以后，<< 会阻止新的元素进来，直到队列中有剩余的空间。完美的解决了上面的问题！

重构

这段代码非常的丑陋，不要在你的应用中使用上面的代码， 稍微抽象一下会使代码更简洁，也就是把完美的业务逻辑从中抽离出去.

感谢关注Ruby程序员公众账号!~