queue implementation using MySQL

enqueue.rb

require 'mysql2-cs-bind'

def get_db
  return Mysql2::Client.new(
    :host => 'localhost',
    :port => 3306,
    :username  => 'root',
    :password  => '',
    :database  => 'queue_test',
    :reconnect => true,
  )
end

def enqueue
  db = get_db
  db.xquery("INSERT INTO `t` (`created_at`, `executed_at`, `target_id`, `updated_at`) VALUES (?, ?, ?, ?)", Time.now, Time.now, 1, Time.now)
  db.close
end

100000.times do
  enqueue
end

receiver.rb

require 'mysql2-cs-bind'

def get_db
  return Mysql2::Client.new(
    :host => 'localhost',
    :port => 3306,
    :username  => 'root',
    :password  => '',
    :database  => 'queue_test',
    :reconnect => true,
  )
end

def dequeue
  db = get_db

  queues = db.query("select * from t where status = 0 and executed_at < now()")
  queues.each do |q|
    db.xquery("update t set status = 1 where id = ? and status = 0", q["id"])
    if db.affected_rows() == 1
      puts q["id"]
      db.xquery("delete from t where id = ?", q['id'])
    end
  end
  db.close
end

10000.times do
  dequeue
end

schema.sql

CREATE TABLE `t` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `target_id` int(11) NOT NULL,
  `status` int(11) NOT NULL DEFAULT '0',
  `executed_at` datetime DEFAULT NULL,
  `created_at` datetime DEFAULT NULL,
  `updated_at` datetime DEFAULT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_targetId` (`target_id`),
  KEY `idx_status_executedAt` (`status`,`executed_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8