rb_producer.rb

require "kafka"
require "json"

class Producer
  def initialize(producer_id)
    @producer_id = producer_id
    @kafka = Kafka.new(["localhost:9092"], client_id: @producer_id)
    @count = 0
  end

  def run
    random = Random.new

    puts "#{@producer_id}: Starting to publish messages..."

    loop do
      @count += 1
      item_id = random.rand(10) + 1
      message = {
        bid_amount: item_id * @count,
        producer_id: @producer_id
      }
      log_and_deliver_message(key: item_id, message: message)
      sleep 1
    end
  end

  private

  def log_and_deliver_message(key:, message:)
    message = JSON.dump(message)
    key = key.to_s
    puts "#{@producer_id}: Publishing message with key=#{key} message=#{message}"
    @kafka.deliver_message(message, key: key, topic: "bids")
  end
end

1.upto(2).map { |id|
  Thread.new do
    Producer.new("producer_#{id}").run
  end
}

gets