Ruby中TCPSocket多线程共享是否线程安全？dup/clone是否可行？

问题描述 #

我正在用Ruby标准库实现一个简易pub/sub系统，客户端可订阅主题接收消息或向主题发布消息，消息会发送给所有订阅者。仅使用标准TCPServer，假设所有消息为单行，通过gets读取。用Ruby的Queue做线程间通信，避免访问非线程安全的共享内存。现在遇到必须在多线程间共享TCPSocket客户端的情况：一个线程循环阻塞等待socket上的新数据，另一个线程等待已发布的新消息并写入该客户端。想问：

1. Ruby中的TCPSocket是否线程安全？
2. 如果不安全，用简单的dup或clone让每个线程持有自己的socket引用是否可行？

代码参考：

require 'socket'

socket = TCPServer.new(4242)

processing_queue = Queue.new
Thread.start({}) do |subscriptions|
  while event = processing_queue.pop
    command, *args = event
    case command
    when 'publish'
      topic, message = args
      subscriptions[topic].each do |subscription_queue|
        subscription_queue.push(message)
      end
    when 'subscribe'
      topic, subscription_queue = args
      subscriptions[topic] = [] if subscriptions[topic].nil?
      subscriptions[topic] << subscription_queue
    end
  end
end

loop do
  Thread.start(socket.accept, Queue.new) do |client, queue|
    writer_queue = Queue.new
    Thread.start do
      while response = writer_queue.pop
        client.puts(response)
      end
    end

    while request = client.gets
      command, *args = request.split(' ')
      case command
      when 'subscribe'
        topic = args[0]
        Thread.start(Queue.new) do |subscription_queue|
          processing_queue << ['subscribe', topic, subscription_queue]
          while message = subscription_queue.pop
            writer_queue << message
          end
        end
        writer_queue << 'OK'
      when 'publish'
        topic = args.shift
        message = args.join(' ')
        processing_queue << ['publish', topic, message]
        writer_queue << 'OK'
      end
    end
    client.close
  end
end

socket.close

回答 #

1. TCPSocket是否线程安全？ #

Ruby的TCPSocket（本质是IO子类）不是线程安全的。底层的socket读写操作没有内置的线程同步机制，多个线程同时对同一个TCPSocket实例进行操作时，可能出现以下问题：

• 读写操作交错，导致数据混乱（比如写入的消息被拆分，或者读取到不完整的内容）
• 资源竞争引发的异常，比如socket状态被意外修改

不过在你的代码场景中，一个线程只负责读（gets），另一个只负责写（puts），这种单方向的分工虽然能降低风险，但依然存在潜在问题——Ruby的IO对象内部状态（比如缓冲区管理）并没有针对多线程访问做保护，极端情况下还是可能出现异常。

2. 使用dup/clone是否可行？ #

使用dup或clone创建TCPSocket的副本是可行的方案，具体原因如下：

• dup会创建一个新的IO对象，共享同一个底层socket文件描述符，但每个副本拥有独立的内部状态（比如读写位置指针、缓冲区）
• 每个线程持有自己的socket副本后，读写操作的内部状态不会互相干扰，能避免多线程共享同一实例带来的风险

修改你的代码也很简单，只需在创建写线程时复制socket：

Thread.start(client.dup) do |client_writer|
  while response = writer_queue.pop
    client_writer.puts(response)
  end
  client_writer.close
end

需要注意的是：

• 虽然副本共享底层socket，但关闭任意一个副本都会关闭底层连接，所以要确保所有线程完成操作后再关闭，或者统一管理关闭时机
• 读写操作本身在系统层面，socket的写操作只要数据不超过内核缓冲区，通常是原子的，但通过dup隔离Ruby层面的IO对象状态，能彻底避免Ruby内部的线程安全问题

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