単一のテーブル vs 複数のテーブル vs 専用システム：MySQL ジョブキューの比較

MySQL でジョブキューを実装する最適な方法

ジョブキューは、タスクやメッセージを非同期に処理するために使用される一般的なパターンです。MySQL は、ジョブキューを実装するための強力なデータベースですが、最適な方法を選択することは重要です。

方法

MySQL でジョブキューを実装する方法はいくつかあります。最も一般的な方法は次のとおりです。

• 単一のテーブルを使用する
  • 最も単純な方法ですが、スケーラビリティとパフォーマンスが制限される可能性があります。
  • メッセージが完了したときにステータスを更新する必要があります。
• 複数のテーブルを使用する
  • スケーラビリティとパフォーマンスを向上させることができます。
  • メッセージの状態を管理するための別のテーブルが必要です。
• 専用のキューシステムを使用する

最適な方法を選択する

最適な方法は、要件によって異なります。次の要素を考慮する必要があります。

• メッセージの量
• 処理速度
• 機能

例

単一のテーブルを使用する

CREATE TABLE jobs (
  id INT PRIMARY KEY AUTO_INCREMENT,
  message TEXT,
  status ENUM('pending', 'processing', 'completed'),
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

CREATE TABLE jobs (
  id INT PRIMARY KEY AUTO_INCREMENT,
  message TEXT
);

CREATE TABLE job_statuses (
  id INT PRIMARY KEY AUTO_INCREMENT,
  job_id INT NOT NULL,
  status ENUM('pending', 'processing', 'completed'),
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  FOREIGN KEY (job_id) REFERENCES jobs(id)
);

専用のキューシステムを使用する

RabbitMQ や Kafka などの専用のキューシステムを使用することができます。

import mysql.connector

# データベースへの接続
db = mysql.connector.connect(
    host="localhost",
    user="root",
    password="password",
    database="job_queue"
)

# ジョブを追加する
def add_job(message):
    cursor = db.cursor()
    cursor.execute("INSERT INTO jobs (message, status) VALUES (%s, 'pending')", (message,))
    db.commit()
    cursor.close()

# ジョブを完了する
def complete_job(job_id):
    cursor = db.cursor()
    cursor.execute("UPDATE jobs SET status = 'completed' WHERE id = %s", (job_id,))
    db.commit()
    cursor.close()

# ジョブを取得する
def get_jobs():
    cursor = db.cursor()
    cursor.execute("SELECT * FROM jobs WHERE status = 'pending'")
    jobs = cursor.fetchall()
    cursor.close()
    return jobs

# ジョブを処理する
def process_job(job):
    # ここにジョブ処理のロジックを記述する
    print(f"Processing job: {job['message']}")
    complete_job(job['id'])

# メインプログラム
while True:
    # ジョブを取得する
    jobs = get_jobs()

    # ジョブを処理する
    for job in jobs:
        process_job(job)

    # 1 秒待つ
    time.sleep(1)

import mysql.connector

# データベースへの接続
db = mysql.connector.connect(
    host="localhost",
    user="root",
    password="password",
    database="job_queue"
)

# ジョブを追加する
def add_job(message):
    cursor = db.cursor()
    cursor.execute("INSERT INTO jobs (message) VALUES (%s)", (message,))
    job_id = cursor.lastrowid
    cursor.execute("INSERT INTO job_statuses (job_id, status) VALUES (%s, 'pending')", (job_id,))
    db.commit()
    cursor.close()
    return job_id

# ジョブを完了する
def complete_job(job_id):
    cursor = db.cursor()
    cursor.execute("UPDATE job_statuses SET status = 'completed' WHERE job_id = %s", (job_id,))
    db.commit()
    cursor.close()

# ジョブを取得する
def get_jobs():
    cursor = db.cursor()
    cursor.execute("SELECT j.*, js.status FROM jobs j JOIN job_statuses js ON j.id = js.job_id WHERE js.status = 'pending'")
    jobs = cursor.fetchall()
    cursor.close()
    return jobs

# ジョブを処理する
def process_job(job):
    # ここにジョブ処理のロジックを記述する
    print(f"Processing job: {job['message']}")
    complete_job(job['id'])

# メインプログラム
while True:
    # ジョブを取得する
    jobs = get_jobs()

    # ジョブを処理する
    for job in jobs:
        process_job(job)

    # 1 秒待つ
    time.sleep(1)

注記

• 上記のコードはあくまで例であり、要件に応じて変更する必要があります。
• エラー処理やロックなどの追加機能を実装する必要があります。

Redis は、インメモリデータ構造ストアであり、メッセージキューに適しています。MySQL よりも高速でスケーラブルですが、永続的なストレージは提供しません。

Celery を使用する

Celery は、タスクキューワーキング用の分散メッセージングシステムです。RabbitMQ や Kafka などのメッセージブローカーと連携して、ジョブの実行を管理します。

Amazon Simple Queue Service (SQS) を使用する

SQS は、クラウドベースのメッセージキューサービスです。スケーラブルで高可用性であり、他の AWS サービスと簡単に統合できます。

• メッセージの量
• 処理速度
• 機能
• コスト
• スキルセット

各方法の比較