Redis訂閱塊中的ActiveJobperform_later

我想使用 ruby?? redis 客戶端訂閱頻道。訂閱成功后，我想使用perform_later. 然后，此作業將執行一些作業并產生將發布到channelRedis 頻道的結果。這會產生以下錯誤：

Processing by Api::LocationController#yyy as */*
  Parameters: {"zip_code"=>"503400"}
[ActiveJob] Failed enqueuing XXXJob to Sidekiq(default): NoMethodError (undefined method `call_pipeline' for #<Redis::SubscribedClient:0x00007fd9222cdf78 @client=#<Redis::Client:0x00007fd921e1b890 @options={:url=>"redis://localhost?db=xxx", :scheme=>"redis", :host=>"localhost", :port=>6379, :path=>nil, :read_timeout=>0, :write_timeout=>5.0, :connect_timeout=>5.0, :timeout=>5.0, :username=>nil, :password=>nil, :db=>0, :driver=>Redis::Connection::Ruby, :id=>nil, :tcp_keepalive=>0, :reconnect_attempts=>1, :reconnect_delay=>0.0, :reconnect_delay_max=>0.5, :inherit_socket=>false, :logger=>nil, :sentinels=>nil, :role=>:master, :_parsed=>true}, @reconnect=true, @logger=nil, @connection=#<Redis::Connection::Ruby:0x00007fd921df08c0 @sock=#<Redis::Connection::TCPSocket:fd 26>>, @command_map={}, @pending_reads=-3, @connector=#<Redis::Client::Connector:0x00007fd921e1a2d8 @options={:url=>"redis://localhost?db=xxx", :scheme=>"redis", :host=>"localhost", :port=>6379, :path=>nil, :read_timeout=>5.0, :write_timeout=>5.0, :connect_timeout=>5.0, :timeout=>5.0, :username=>nil, :password=>nil, :db=>0, :driver=>Redis::Connection::Ruby, :id=>nil, :tcp_keepalive=>0, :reconnect_attempts=>1, :reconnect_delay=>0.0, :reconnect_delay_max=>0.5, :inherit_socket=>false, :logger=>nil, :sentinels=>nil, :role=>:master, :_parsed=>true}>, @pid=54943>>)
Completed 500 Internal Server Error in 26ms (Allocations: 2374)


  
NoMethodError (undefined method `call_pipeline' for #<Redis::SubscribedClient:0x00007fd9222cdf78 @client=#<Redis::Client:0x00007fd921e1b890 @options={:url=>"redis://localhost?db=xxx", :scheme=>"redis", :host=>"localhost", :port=>6379, :path=>nil, :read_timeout=>5.0, :write_timeout=>5.0, :connect_timeout=>5.0, :timeout=>5.0, :username=>nil, :password=>nil, :db=>0, :driver=>Redis::Connection::Ruby, :id=>nil, :tcp_keepalive=>0, :reconnect_attempts=>1, :reconnect_delay=>0.0, :reconnect_delay_max=>0.5, :inherit_socket=>false, :logger=>nil, :sentinels=>nil, :role=>:master, :_parsed=>true}, @reconnect=true, @logger=nil, @connection=#<Redis::Connection::Ruby:0x00007fd921df08c0 @sock=nil>, @command_map={}, @pending_reads=-3, @connector=#<Redis::Client::Connector:0x00007fd921e1a2d8 @options={:url=>"redis://localhost?db=xxx", :scheme=>"redis", :host=>"localhost", :port=>6379, :path=>nil, :read_timeout=>5.0, :write_timeout=>5.0, :connect_timeout=>5.0, :timeout=>5.0, :username=>nil, :password=>nil, :db=>0, :driver=>Redis::Connection::Ruby, :id=>nil, :tcp_keepalive=>0, :reconnect_attempts=>1, :reconnect_delay=>0.0, :reconnect_delay_max=>0.5, :inherit_socket=>false, :logger=>nil, :sentinels=>nil, :role=>:master, :_parsed=>true}>, @pid=54943>>):

  
app/controllers/api/location_controller.rb:10:in `block (3 levels) in yyy'
app/controllers/api/location_controller.rb:8:in `block in yyy'
app/controllers/api/location_controller.rb:5:in `yyy'

該yyy函式是在GET請求時呼叫的控制器的一部分：

  def yyy
    REDIS_POOL.with { |r|
      data = r.get("key")
      if data.nil?
        r.subscribe("channel") { |on|
          on.subscribe {
            XXXJob.perform_later(params)
          }
          on.message { |_, message|
            puts message
            render json: message
            r.unsubscribe("channel")
          }
        }
      else
        render json: data
      end
    }
  end

在我看來，Redissubscribe塊丟失了一些背景關系，這對perform_later函式很重要。如果我將XXXJob.perform_later(params)呼叫移到呼叫上方r.subscribe("channel")并因此移到塊之外，則相同的代碼有效。但是，我想確保該作業僅在訂閱 Redis 頻道成功后才開始。有沒有辦法實作這一目標？

uj5u.com熱心網友回復：

它很難。它不會作業，因為 Sidekiq 的默認連接池總是重用執行緒的當前連接，它正在處理 subscribe 呼叫并且該呼叫是不可重入的。您需要手動創建一個單獨的連接池并直接使用它。

ANOTHER_POOL = ConnectionPool.new { Redis.new }

on.subscribe do
  Sidekiq::Client.via(POOL) do
    SomeWorker.perform_async(1,2,3)
    SomeOtherWorker.perform_async(1,2,3)
  end
end

標籤：红宝石轨道 红宝石 Redis 边基克 rails-activejob

上一篇：Array#productRangeError:太大而無法生成

下一篇：在Ruby中從Github讀取私有檔案