如何从外部与EventMachine线程内的ActionCableClient交互并访问数据？

嘿，我完全懂你现在的困惑——EventMachine的事件驱动模型和普通Ruby线程的交互确实有点绕，尤其是你刚接触这块的时候。咱们直接针对你的两个核心问题来拆解解决方案：

1. 从外部触发EventMachine内部的ActionCable任务 #

直接尝试从外部线程访问my_thread里的client实例是行不通的：一来线程内部的局部变量外部根本拿不到，二来跨线程直接操作EventMachine上下文里的对象会有严重的线程安全问题。

最稳妥的方式是用线程安全的消息队列来传递指令：外部线程把要执行的动作（比如调用perform）放进队列，然后在EventMachine的事件循环里定期检查队列，执行对应的操作。

2. 安全访问EventMachine内的变量 #

你用的@@message_storage是类变量，属于全局共享，线程环境下很容易出现竞态条件（比如多个线程同时读写导致数据错乱）。推荐用线程安全的数据结构，比如Concurrent::Array（需要安装concurrent-ruby gem），或者用普通Array配合Mutex来保护读写操作。

重构后的完整代码示例 #

require 'action_cable_client'
require 'concurrent-ruby' # 需要先安装：gem install concurrent-ruby

# 初始化线程安全的组件
action_queue = Queue.new
message_storage = Concurrent::Array.new

my_thread = Thread.new do
  EventMachine.run do
    uri = "ws://127.0.0.1:3000/cable"
    client = ActionCableClient.new(uri, 'LoggingChannel')

    client.connected do
      puts 'successfully connected.'
    end

    client.disconnected do
      puts 'disconnected.'
    end

    client.received do |message|
      puts "Got a message: #{message}"
      message_storage << message # Concurrent::Array是线程安全的，直接加就行
    end

    # 定期检查队列，处理外部发来的动作请求
    EM.add_periodic_timer(0.1) do
      until action_queue.empty?
        action, params = action_queue.pop
        case action
        when :perform
          action_name, data = params
          client.perform(action_name, data)
        end
      end
    end
  end
end

# 主应用逻辑
def some_actions; end

some_actions()

if error_condition # 替换成你的实际错误判断
  # 往队列里放perform指令
  action_queue << [:perform, ['error', {message: 'error 123 occured'}]]
end

if warning_condition # 替换成你的实际警告判断
  action_queue << [:perform, ['info', {message: 'some information for you'}]]
end

# 安全访问消息存储
puts "Currently there are #{message_storage.size} messages in storage"

# 记得在程序退出时优雅关闭EventMachine和线程
at_exit do
  EventMachine.stop if EventMachine.reactor_running?
  my_thread.join
end

关键细节说明 #

• 消息队列action_queue：Ruby标准库的Queue是线程安全的，外部线程用<<添加任务，EventMachine线程用pop取出执行，完全不用担心竞态问题。
• Concurrent::Array：替代普通Array，所有读写操作都是线程安全的，不用手动加锁。如果不想额外装gem，也可以用普通Array加Mutex：

  message_storage = []
  storage_mutex = Mutex.new
  
  # 写入时
  storage_mutex.synchronize { message_storage << message }
  
  # 读取时
  storage_mutex.synchronize { puts message_storage.size }
  

• EM.add_periodic_timer：每隔0.1秒检查一次队列，这个间隔可以根据你的需求调整，既保证响应速度，又不会太占用资源。

这样就能安全地从外部线程和EventMachine里的ActionCable客户端交互，同时也能安全访问内部的消息存储啦！

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