在不同的线程中运行 websocket onmessage

Question

我在如下服务器上使用 websocket。它响应onmessage 事件，并根据消息条件执行不同的任务：

require "websocket-eventmachine-server"

WebSocket::EventMachine::Server.start(host: some_server_name, port: some_port) do |ws|
  # (A) Here, the current thread is the main thread
  ws.onmessage do |s|
    if foo
      # (B) Here, the current thread is the main thread
      ...
    else
      # (C) Here, the current thread is the main thread
      ...
    end 
  end
end

每个onmessage事件执行的线程（上面描述为B和C）每次都是相同的，它们与主线程相同（上面描述为A）。

我想以C 的形式在单独的线程中执行B 中的代码。一种方法是将B 和C 中的操作放在一个新线程中，如下所示：

WebSocket::EventMachine::Server.start(host: some_server_name, port: some_port) do |ws|
  # (A) Here, the current thread is the main thread
  ws.onmessage do |s|
    if foo
      # (B) Here, the current thread will be created each time.
      Thread.new{...}
    else
      # (C) Here, the current thread will be created each time.
      Thread.new{...}
    end
  end
end

但是每次发生事件时创建一个新线程似乎很繁重，并且使响应变慢。所以，我希望在B 中处理的所有onmessage 事件之间共享一个线程，在C 中处理的所有事件之间共享另一个线程：

WebSocket::EventMachine::Server.start(host: some_server_name, port: some_port) do |ws|
  # (A) Here, the current thread is the main thread
  ws.onmessage do |s|
    if foo
      # (B) I want this part to be executed in a thread
      #     that does not change each time, but is different from the thread in C
      ...
    else
      # (C) I want this part to be executed in a thread
      #     that does not change each time, but is different from the thread in B
      ...
    end
  end
end

有什么好的方法可以做到这一点？或者，是否有更好的结构来以相互非阻塞的方式响应 websocket onmessage 事件？

Answer 1

使用EventMachine.defer方法在其内部线程池中执行代码。

Answer 2

您可以将接收到的消息制作成队列，并按队列制作一个线程以执行 特点：

def do_foo(message)
   .... your code
end
def do_fee(message)
   .... your code
end

queueA= Queue.new
queueB= Queue.new
Thread.new { loop { do_foo(queueA.pop) } }
Thread.new { loop { do_fee(queueB.pop) } }
WebSocket::EventMachine::Server.start(host: some_server_name, port: some_port) do |ws|
  # (A) Here, the current thread is the main thread
  ws.onmessage do |s|
    if foo
       queueA.push(s)
    else
       queueB.push(s)
    end
  end
end

警告！如果 do_foo/fee 需要在 websocket 上发送消息，你应该 在 EM.next_tick { if .. } 中调用 'if foo..'。