在末尾更新编辑：显示工作代码。除调试代码外，未修改主模块。注意：我确实遇到了我已经提到的有关终止之前需要退订的问题。
该代码看起来正确。我想看看您如何实例化它。
在config / application.rb中，您可能至少有类似以下内容：
require 'ws_communication'config.middleware.use WsCommunication
然后，在您的Javascript客户端中，您应该具有以下内容：
var ws = new WebSocket(uri);
您是否实例化WsCommunication的另一个实例？这会将@clients设置为一个空数组，并且可能会出现您的症状。这样的事情是不正确的：
var ws = new WsCommunication;
如果您显示客户端，可能会对我们有所帮助，如果此帖子对您没有帮助，也可能对config / application.rb有所帮助。
顺便说一句，我同意这样的意见，即@clients在任何更新上都应该由互斥体保护，即使不能读取也是如此。这是一个动态结构，可以在事件驱动的系统中随时更改。
redis-mutex是一个不错的选择。（希望该链接是正确的，因为目前Github似乎在所有内容上抛出500个错误。）
您可能还注意到$ redis.publish返回接收消息的客户端数量的整数值。
最后，您可能会发现需要在终止之前确保取消订阅频道。我遇到过这样的情况，由于没有清除同一频道的早期订阅，我最终多次，甚至多次发送每条消息。由于您正在订阅一个线程中的通道，因此您将需要在同一线程中退订，否则进程将“挂起”，等待正确的线程神奇地出现。我通过设置“取消订阅”标志然后发送消息来处理这种情况。然后，在on.message块中，我测试取消订阅标志并在那里发出取消订阅。
您提供的模块，仅进行了少量调试修改：
require 'faye/websocket'require 'redis'class WsCommunication KEEPALIVE_TIME = 15 #seconds CHANNEL = 'vip-deck' def initialize(app) @app = app @clients = [] uri = URI.parse(ENV['REDISCLOUD_URL']) $redis = Redis.new(host: uri.host, port: uri.port, password: uri.password) Thread.new do redis_sub = Redis.new(host: uri.host, port: uri.port, password: uri.password) redis_sub.subscribe(CHANNEL) do |on| on.message do |channel, msg| puts "Message event. Clients receiving:#{@clients.count};" @clients.each { |ws| ws.send(msg) } end end end end def call(env) if Faye::WebSocket.websocket?(env) ws = Faye::WebSocket.new(env, nil, {ping: KEEPALIVE_TIME}) ws.on :open do |event| @clients << ws puts "Open event. Clients open:#{@clients.count};" end ws.on :message do |event| receivers = $redis.publish(CHANNEL, event.data) puts "Message published:#{event.data}; Receivers:#{receivers};" end ws.on :close do |event| @clients.delete(ws) puts "Close event. Clients open:#{@clients.count};" ws = nil end ws.rack_response else @app.call(env) end endend我提供的测试订户代码：
# encoding: UTF-8puts "Starting client-subscriber.rb"$:.unshift File.expand_path '../lib', File.dirname(__FILE__)require 'rubygems'require 'eventmachine'require 'websocket-client-simple'puts "websocket-client-simple v#{WebSocket::Client::Simple::VERSION}"url = ARGV.shift || 'ws://localhost:3000'EM.run do ws = WebSocket::Client::Simple.connect url ws.on :message do |msg| puts msg end ws.on :open do puts "-- Subscriber open (#{ws.url})" end ws.on :close do |e| puts "-- Subscriber close (#{e.inspect})" exit 1 end ws.on :error do |e| puts "-- Subscriber error (#{e.inspect})" endend我提供的测试发布者代码。发布者和订阅者可以很容易地合并，因为这只是测试：
# encoding: UTF-8puts "Starting client-publisher.rb"$:.unshift File.expand_path '../lib', File.dirname(__FILE__)require 'rubygems'require 'eventmachine'require 'json'require 'websocket-client-simple'puts "websocket-client-simple v#{WebSocket::Client::Simple::VERSION}"url = ARGV.shift || 'ws://localhost:3000'EM.run do count ||= 0 timer = EventMachine.add_periodic_timer(5+rand(5)) do count += 1 send({"MESSAGE": "COUNT:#{count};"}) end @ws = WebSocket::Client::Simple.connect url @ws.on :message do |msg| puts msg end @ws.on :open do puts "-- Publisher open" end @ws.on :close do |e| puts "-- Publisher close (#{e.inspect})" exit 1 end @ws.on :error do |e| puts "-- Publisher error (#{e.inspect})" @ws.close end def self.send message payload = message.is_a?(Hash) ? message : {payload: message} @ws.send(payload.to_json) endend在机架中间件层运行所有这些的示例config.ru：
require './controllers/main'require './middlewares/ws_communication'use WsCommunicationrun Main.new
这是主要的。我从运行版本中剥离了它，因此如果您使用它，可能需要对其进行调整：
%w(rubygems bundler sinatra/base json erb).each { |m| require m }ENV['RACK_ENV'] ||= 'development'Bundler.require$: << File.expand_path('../', __FILE__)$: << File.expand_path('../lib', __FILE__)Dir["./lib*.rb"].each { |file| require file }env = ENV['OS'] == 'Windows_NT' ? 'development' : ENV['RACK_ENV'] class Main < Sinatra::base env = ENV['OS'] == 'Windows_NT' ? 'development' : ENV['RACK_ENV'] get "/" do erb :"index.html" end get "/assets/js/application.js" do content_type :js @scheme = env == "production" ? "wss://" : "ws://" erb :"application.js" end end
