Ruby是否提供原生双向字符串缓冲类？求线程间通信实现方案

Ruby标准库并没有提供原生的双向字符串缓冲类，但你完全可以不用mkfifo就实现线程间的这种流式通信需求，下面给你两种实用的方案：

方案一：用Ruby标准库的Queue实现生产者-消费者模式 #

Queue是Ruby为线程安全通信设计的原生类，完美适配你这种生产者线程写入、消费者线程读取的场景，用法简单且无需自己处理同步问题。

改造你的代码示例如下：

require 'thread'

queue = Queue.new

# 生产者线程
Thread.new do
  loop do
    # 模拟生成业务结果
    result = "Processed data #{rand(100)}"
    queue << result
    sleep 0.5 # 模拟工作耗时
  rescue Exception => e
    queue.close
    raise
  end
end

# 消费者线程
Thread.new do
  loop do
    begin
      # 阻塞等待获取结果，队列关闭后会抛出ThreadError
      result = queue.pop
      puts "Received: #{result}"
    rescue ThreadError
      puts "Queue closed, exiting consumer thread"
      break
    end
  end
end.join

关键细节： #

• Queue#pop会自动阻塞直到有元素可用，无需手动轮询；
• 队列关闭后调用pop会抛出ThreadError，捕获这个异常即可优雅终止消费者循环；
• 也可以在生产者结束时推入一个特殊标记（比如nil），消费者检测到标记后主动退出，这种方式更可控。

方案二：自定义基于StringIO和互斥锁的双向缓冲类 #

如果你希望接口更贴近你最初设想的Buffer类（比如支持readline、eof?这类IO风格方法），可以自己封装一个线程安全的缓冲类，用Mutex同步StringIO的读写操作：

require 'stringio'
require 'thread'

class BidirectionalBuffer
  def initialize
    @io = StringIO.new
    @mutex = Mutex.new
    @closed = false
    @condition = ConditionVariable.new
  end

  # 写入数据（自动追加换行符适配readline）
  def <<(data)
    @mutex.synchronize do
      raise IOError, "Buffer is closed" if @closed
      @io.write("#{data}\n")
      @condition.signal # 通知等待读取的线程
    end
    self
  end

  # 读取一行数据
  def readline
    @mutex.synchronize do
      # 无数据时阻塞，直到有新数据或缓冲区关闭
      while @io.pos >= @io.size && !@closed
        @condition.wait(@mutex)
      end

      raise EOFError if eof?

      # 重置指针并读取一行，同时保留剩余未读数据
      @io.rewind
      line = @io.readline
      remaining_data = @io.read
      @io.reopen(remaining_data || "")
      line
    end
  end

  def close
    @mutex.synchronize do
      @closed = true
      @condition.signal # 唤醒所有等待的线程
    end
  end

  def eof?
    @mutex.synchronize do
      @closed && @io.pos >= @io.size
    end
  end
end

使用这个类的示例代码和你最初的需求几乎一致：

buf = BidirectionalBuffer.new

Thread.new do
  5.times do |i|
    result = "Result #{i+1}"
    buf << result
    sleep 0.5
  end
  buf.close
end

Thread.new do
  until buf.eof?
    begin
      line = buf.readline
      puts "Read: #{line.chomp}"
    rescue EOFError
      break
    end
  end
  puts "Buffer closed, done reading"
end.join

关键细节： #

• 通过Mutex保证读写操作的原子性，避免多线程下的数据竞争；
• 用ConditionVariable让消费者线程在无数据时阻塞，避免空轮询浪费资源；
• 实现了eof?方法，完全匹配你最初的逻辑判断需求。

方案选择建议 #

• 如果只是需要简单的线程间消息传递，Queue是最省心的选择，不用自己造轮子；
• 如果需要严格贴合IO流的接口风格，自定义BidirectionalBuffer类能更好地满足你的需求。

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