nsqd-ruby debugger

output.sh

wyatt@mac:nsq/src $ ruby producer.rb
SizeQueue: 2495
SizeQueue: 4947
SizeQueue: 7446
SizeQueue: 9999
SizeQueue: 9891
SizeQueue: 10000
SizeQueue: 10000
SizeQueue: 10000
SizeQueue: 9916
SizeQueue: 9752
SizeQueue: 10000
SizeQueue: 9957
SizeQueue: 9999
SizeQueue: 9998
SizeQueue: 9985
SizeQueue: 9999
SizeQueue: 10000
SizeQueue: 9967
SizeQueue: 9998
SizeQueue: 10000
#<Thread:0x00007fb2c207f0b8> - 4.399s deals: 100000 SizeQueue: 9998
connected?: true, SizeQueue: 9995
terminate: ["127.0.0.1:4150", "127.0.0.1:4250"], SizeQueue: 0
connected?: false. passed time: 0.842s
end..

producer.rb

require 'nsq'

#Nsq.logger = Logger.new(STDOUT)

MSG = %({"job_class":"Async::RemindNotifyJob","job_id":"50c95524-3514-4304-8086-da54e4ef6a71","provider_job_id":null,"queue_name":"eca_remind_notify","priority":null,"arguments":["df595f39-73e6-46b7-a853-3653bfe45fe2"],"executions":0,"locale":"zh-CN"})

class Nsq::Connection
    # closes the connection and stops listening for messages
    def close_connection
      cls if connected?
      stop_write_loop
      stop_read_loop
      # the problem is here. 
      # 1. the code running here, the SizedQueue.size is 0
      # 2. the msg is also all write to socket
      # 3. but is i comment `sleep 1` the Message size in nsqd will randomly missing. 
      sleep 1
      @socket.close if @socket
      @socket = nil
      @connected = false
    end

    def write_to_socket(raw)
      # redirect it`s output to file
      puts raw
      @socket.write(raw)
    end
end



def push_Nw_msg(n)
  p = Nsq::Producer.new(nsqlookupd: '127.0.0.1:4161', topic: 'st')
  # need time to connect to nsqlookupd
  sleep 1
  t = Time.now
  sum = 0
  q = nil
  1.upto(n * 1_0000) do |i|
    sum += 1
    q = p.write(MSG)
    if i % 5000 == 0
      puts "SizeQueue: #{q.size}"
    end
  end
  puts "#{Thread.current} - #{(Time.now - t).round(3)}s deals: #{sum} SizeQueue: #{q.size}"
  puts "connected?: #{p.connected?}, SizeQueue: #{q.size}"
  t2 = Time.now
  puts "terminate: #{p.terminate}, SizeQueue: #{q.size}"
  puts "connected?: #{p.connected?}. passed time: #{(Time.now - t2).round(3)}s"
end

#ts = []
#1.upto(1) do
# ts << Thread.new { push_Nw_msg(10) }
#end

#ts.map(&:join)
push_Nw_msg(10)
puts 'end..'