如何在websocket eventmachine中断开redis客户端

Question

我正在尝试构建一个 websocket 服务器，每个客户端都建立自己的用于发布和订阅的 redis 连接。

当redis服务器运行时，当客户端连接到websocket服务器时，我可以看到两个新连接正在建立，我也可以向客户端发布数据，但是当客户端断开与websocket服务器的连接时，我也想要断开与 Redis 的连接。我该怎么做？

也许我做错了，但这是我的代码。

#require 'redis'
require 'em-websocket'
require 'em-hiredis'
require 'json'

CLIENTS = Hash.new

class PubSub
  def initialize(client)
    @socket = client.ws
    # These clients can only be used for pub sub commands
    @publisher = EM::Hiredis.connect #Later I will like to disconnect this
    @subscriber = EM::Hiredis.connect #Later I will like to disconnect this
    client.connections << @publisher << @subscriber
  end
  def subscribe(channel)
    @channel = channel
    @subscriber.subscribe(channel)
    @subscriber.on(:message) { |chan, message|
      @socket.send message
    }
  end
  def publish(channel,msg)
    @publisher.publish(channel, msg).errback { |e|
      puts [:publisherror, e]
    }
  end
  def unsubscribe()
    @subscriber.unsubscribe(@channel)
  end
end

class Client
  attr_accessor :connections, :ws
  def initialize(ws)
    @connections = []
    @ws = ws
  end
end

EventMachine.run do
  # Creates a websocket listener
  EventMachine::WebSocket.start(:host => '0.0.0.0', :port => 8081) do |ws|

    ws.onopen do
      # I instantiated above
      puts 'CLient connected. Creating socket'      
      @client = Client.new(ws)
      CLIENTS[ws] = @client
    end

    ws.onclose do
      # Upon the close of the connection I remove it from my list of running sockets
      puts 'Client disconnected. Closing socket'

      @client.connections.each do |con|
        #do something to disconnect from redis
      end

      CLIENTS.delete ws
    end

    ws.onmessage { |msg|
      puts "Received message: #{msg}"
      result = JSON.parse(msg)
      if result.has_key? 'channel'
        ps = PubSub.new(@client)
        ps.subscribe(result['channel'])
      elsif result.has_key? 'publish'
         ps = PubSub.new(ws)
         ps.publish(result['publish']['channel'],result['publish']['msg']);
      end
    }
  end
end

Answer 1

此版本的em-hiredis支持关闭连接：https://github.com/whatupdave/em-hiredis

Answer 2

这是我会（并且做过很多次）的方式： 而不是总是为每个客户端打开和关闭连接，您可以保持每个线程/光纤打开 1 个连接，具体取决于您的并发性所基于的内容，这样，如果您使用线程/光纤的轮询，一旦它们中的每一个都有其连接他们会保留并重复使用它们。

直到现在，我并没有在 websocket 上做太多工作（我一直在等待一个标准的实现），但我相信你也可以将这种想法应用到它上。

你也可以做什么 rails/activerecord: keeo 一个 redis 连接池，每次你需要使用你请求一个连接时，使用它并释放它，它可能看起来像这样：

def handle_request(request)
  @redis_pool.get_connection do |c|
    # [...]
  end
end

在产生块之前，从可用的连接中获取连接，之后连接被标记为空闲。

Answer 3

这已添加到 em-hiredis：https://github.com/mloughran/em-hiredis/pull/6