运维开发网

ruby-on-rails – 访问rails线程中的变量

运维开发网 https://www.qedev.com 2020-07-23 18:00 出处:网络 作者:运维开发网整理
我正在构建一个基于网络的幻灯片放映的应用程序,其中一个“主”用户可以在幻灯片之间移动,每个人的浏览器都会跟随.为此,我使用websockets和Redis作为全局通道来发送消息.每个连接的客户端都有一个存储在数组中的信息@clients. 然后我有一个单独的线程用于订阅Redis通道,其中定义了一个’on.message’块,它应该向@clients数组中的每个人发送一条消息,但该块内的该数组是空
我正在构建一个基于网络的幻灯片放映的应用程序,其中一个“主”用户可以在幻灯片之间移动,每个人的浏览器都会跟随.为此,我使用websockets和Redis作为全局通道来发送消息.每个连接的客户端都有一个存储在数组中的信息@clients.

然后我有一个单独的线程用于订阅Redis通道,其中定义了一个’on.message’块,它应该向@clients数组中的每个人发送一条消息,但该块内的该数组是空的(在任何地方都不是空的)否则在模块中).

几乎遵循这个例子:

https://devcenter.heroku.com/articles/ruby-websockets

相关代码,位于自定义中间件类中:

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'])
    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 @clients.count
          ### prints '0,' no clients receive msg
          @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 @clients.count
      ### prints actual number of clients
    end

    ws.on :message do |event|
      $redis.publish(CHANNEL, event.data)
    end

    ws.on :close do |event|
      @clients.delete(ws)
      ws = nil
    end

    ws.rack_response
  else
    @app.call(env)
  end
end
end

在新线程中访问@clients数组是否为空,因为实例变量不是跨线程共享的?如果是这样,我如何跨线程共享变量?

我也尝试过使用$clients(全局变量,应该可以跨线程访问),但无济于事.

编辑:

男人这个网站充满了抓地力.没有人回答他们只是进行超级小编辑以获得代表.

更新后的编辑:显示工作代码.除调试代码外,主模块未修改.注意:我确实遇到过关于在终止之前需要取消订阅的问题.

代码看起来正确.我想看看你是如何实例化它的.

在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块中,我测试unsubscribe标志并在那里发出取消订阅.

您提供的模块,只需进行少量调试修改:

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
  end
end

我提供的测试用户代码:

# encoding: UTF-8
puts "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})"
  end

end

我提供的测试发布者代码.发布者和订阅者可以轻松组合,因为这些只是测试:

# encoding: UTF-8
puts "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)
  end
end

一个示例config.ru,它在机架中间件层运行所有这些:

require './controllers/main'
require './middlewares/ws_communication'
use WsCommunication
run 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", "./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

扫码领视频副本.gif

0

精彩评论

暂无评论...
验证码 换一张
取 消

关注公众号