EventMachine 和 em-websocket - 从队列中读取并推送到通道

Posted 2023-02-16

【中文标题】EventMachine 和 em-websocket - 从队列中读取并推送到通道

【英文标题】：EventMachine and em-websocket - reading from a queue and pushing to a channel

【发布时间】：2012-10-31 15:55:09

【问题描述】：

我正在使用 eventmachine 读取 HornetQ 主题，推送到由 EM websocket 连接订阅的频道。我需要防止 @topic.receive 循环阻塞，因此创建了一个 proc 并在没有回调的情况下调用 EventMachine.defer。这将无限期地运行。这工作正常。我也可以只使用 Thread.new。

我的问题是，这是从流/队列中读取数据并将数据传递到通道的正确方法吗？还有更好的/其他方法吗？

require 'em-websocket'
require 'torquebox-messaging'

class WebsocketServer

  def initialize
    @channel = EM::Channel.new

    @topic = TorqueBox::Messaging::Topic.new('/topics/mytopic')
  end 

  def start
    EventMachine.run do

      topic_to_channel = proc do
        while true
          msg = @topic.receive
          @channel.push msg
        end
      end

      EventMachine.defer(topic_to_channel)

      EventMachine::WebSocket.start(:host => "127.0.0.1", :port => 8081, :debug => false) do |connection|
        connection.onopen do

          sid = @channel.subscribe  |msg| connection.send msg 

          connection.onclose do            
            @channel.unsubscribe(sid)
          end
        end
      end                  
    end
  end
end

WebsocketServer.new.start

【参考方案1】：

这没关系，但是 EM.defer 会产生 20 个线程，所以我会在您的用例中避​​免使用它。一般来说，我会完全避免使用 EM，尤其是 Java 反应器，因为我们从未完成它。

Torquebox 有一个原生 stomp over websockets 解决方案，在这种情况下这将是一个更好的方法，并为您解决了许多其他封装挑战。

如果你真的想为此坚持使用 EM，那么我会使用 Thread.new 而不是 defer，以避免 19 个空闲线程无缘无故地占用额外的内存。

【讨论】：

我认为 em 用于这里的 websocket 实现，是否有扭矩箱替代方案？