【ruby on rails 5】rabbitMQをrailsアプリで非同期で使う

railsでrabbitMQを使った際に挙動が思った通りにならなかったので質問させていただきます.

環境

rails 5.2.0
ruby 2.5.1
ActiveJob
Sidekiq
rabbitMQ
redis-server

実現したいこと

rabbitMQにmodelの全データを送る.コードはjobで作成し、taskで実行する.（詳細は以下のコード参照）
rabbitMQからデータを受け取るworkerを非同期で複数待機させておき、rabbitMQ内のデータを逐次処理して行く（今回は内部データを表示させるのみだが、時間がかかる処理であることを想定して一つの処理あたり１秒sleepさせる）

コードはjobで作成し、taskで実行する.
一つのworkerが処理を行なっている間は、もう片方のworkerに処理が行くようにする

Code

送信側

ruby
1#app/job/send_jobs.rb
2
3class SendJob < ApplicationJob
4  queue_as :test
5
6  def perform(message)
7
8    connection=set_connection()
9    connection.start
10
11    channel=connection.create_channel
12
13    queue=channel.queue('mss')
14
15    item_array=["message1","message2","message3"]
16
17    item_array.each do |item|
18        queue.publish("#{item}", persistent: true)
19    end
20
21    puts "send #{item_array.count} items"
22  end
23
24  def set_connection
25      connection=Bunny.new(
26          :hostname=>host_name,
27          :port=>port_number,
28          :vhost=>vhost,
29          :user=>user,
30          :pass=>password,
31          automatically_recover: false
32      
33      )
34
35      return connection
36  end
37end

受信側

ruby
1#app/jobs/receive_jobs.rb
2
3class ReceiveJob < ApplicationJob
4
5  queue_as :default
6
7  def perform(id)
8    connection=set_connection()
9
10    connection.start
11
12    channel = connection.create_channel
13    queue = channel.queue('mss')
14    
15    item_asin=Inventory.new()
16
17    puts "[worker#{id}] waiting items"
18    queue.subscribe(block: true) do |_delivery_info, _properties, body|
19         p "[woker#{id}]start checking #{body}"
20         sleep 1
21     end
22
23  end
24
25  def set_connection
26      connection=Bunny.new(
27          :hostname=>host_name,
28          :port=>port_number,
29          :vhost=>vhost,
30          :user=>user,
31          :pass=>password,
32          automatically_recover: false
33      
34      )
35
36      return connection
37  end
38
39end
40

タスク

ruby
1namespace :test do
2
3    task :send => :environment do
4        SendJob.perform_later("test")
5    end
6
7    task :receive => :environment do
8        ReceiveJob.perform_later(1)
9        ReceiveJob.perform_later(2)
10    end
11end

エラー（思い通りにいかなかったこと）

ここからはどの順でプログラムを動かすかによって挙動が違ったので、それぞれのパターンについて述べようと思います。

send task実行後にreceive taskを実行した場合

send taskは正常に実行され、rabbitMQにもデータが行っている（sidekiqおよびrabbitMQのダッシュボードで確認）
worker1,worker2ともに正常に実行されている模様（sidekiqのコンソール上に”[worker 1]waiting for items…”、”[worker 2]waiting for items…”が出ることから確認）
worker2では正常に処理が行われるが、worker1では全く行われていない（worker1に関するコメントが全く出ない）