在Ruby中可以使用IO.pipe进行线程间通信吗？

Question

在Ruby的 IO.pipe 文档中的例子，一个消息是通过进程传递的。

我想做一些类似的事情，但有两个区别。

1. 用线程代替进程
2. 用管子来传递消息，而不是一次性的消息。

这是显而易见的，但不工作的代码。

rd, wr = IO.pipe

reader_thread = Thread.new(rd) do |rd|
  data_received = rd.read
  puts "Read: #{data_received.inspect}"
end

write_thread = Thread.new(wr) do |wr|
  wr.write "Message"
  wr.flush
end

write_thread.join
reader_thread.join

这将导致 reader_thread 牵挂 rd.read.

我可以通过使用 IO#read_nonblock:

reader_thread = Thread.new(rd) do |rd|
  data_received = \
    begin
      rd.read_nonblock(100)
    rescue IO::WaitReadable, IO::EAGAINWaitReadable
      IO.select([rd])
      retry
    end

  puts "Read: #{data_received.inspect}"
end

这是正确的模式吗？还是使用 IO.pipe 线程消息传递的错误工具？

Answer 1

你也可以使用一个 Queue 用于在多个线程之间安全地交换信息。

q = Queue.new

reader_thread = Thread.new(q) do |q|
  data_received = q.pop
  puts "Read: #{data_received.inspect}"
end

write_thread = Thread.new(q) do |q|
  q.push "Message"
end

write_thread.join
reader_thread.join