非同期処理の設計：同期との使い分け基準とジョブキューのリトライ設計

非同期処理とは、依頼を受け付けた時点でいったん応答を返し、処理の本体を後から別のプロセス（ワーカー）で行う方式です。 設計で決めることは「どの処理を非同期にするか」と「うまくいかなかったときにどう再試行し、どこへ退避するか」の2つに集約できます。

この記事では、同期・非同期・定期バッチの選び方を判断表とフロー図で示したうえで、Python と SQLite だけで動く最小のジョブキューを作り、再試行・退避・二重実行が起きる様子を実行結果で確かめます。最後に、そのまま使えるジョブ定義書のテンプレートを記入例つきで載せます。

確認環境（確認日 2026-10-05）：Ubuntu 24.04、Python 3.11.15（標準ライブラリの sqlite3、SQLite 3.45.1）、PostgreSQL 16.15、psycopg 3.3.6。図は Mermaid CLI（@mermaid-js/mermaid-cli 12.0.0）で描画できることを確認しています。

この記事の結論

• 次の3つのどれかに当てはまる処理は非同期にします。 利用者が結果をその場で必要としない処理、数秒を超える処理、外部への呼び出しを含む処理です。
• 起点が「時刻」で、まとめて処理してよいものは定期バッチにします。それ以外で、その場で結果が要る短い処理は同期のままにします。
• 非同期にした処理は同じジョブが2回動くことがある前提で設計します。そのためにリトライ（再試行）と冪等性（何回実行しても結果が同じになる性質）をセットで決めます。
• ジョブごとに「最大試行回数・待ち時間・タイムアウト・冪等キー・退避先・通知先」を定義書に書いておくと、運用で迷いません。

非同期処理にするかどうかの早見表

「数秒」は目安の例です。画面の応答時間の目標が決まっている場合は、その値を基準にしてください。

同期・非同期・定期バッチを選ぶフロー

質問に上から答えていくと、方式が1つに決まります。

flowchart TD
    S([処理を1つ選ぶ]) --> Q1{起点は時刻で<br>まとめて処理してよいか}
    Q1 -- はい --> B[定期バッチ]
    Q1 -- いいえ --> Q2{利用者は結果を<br>その場で必要とするか}
    Q2 -- いいえ --> A[非同期<br>ジョブキューに登録]
    Q2 -- はい --> Q3{数秒以内に終わるか}
    Q3 -- いいえ --> A2[非同期<br>受付番号を返し<br>結果は後から確認]
    Q3 -- はい --> Q4{外部への呼び出しを<br>含むか}
    Q4 -- いいえ --> SY[同期のまま]
    Q4 -- はい --> Q5{外部が止まっても<br>受付は続けたいか}
    Q5 -- はい --> A2
    Q5 -- いいえ --> SY2[同期<br>タイムアウトを設定]

結果がその場で必要なのに時間がかかる処理は、「受付だけを同期で返し、本体を非同期にする」形に分けます。画面には受付番号と状態（受付済み・処理中・完了）を表示し、完了は通知や画面の再読み込みで伝えます。

3方式の違い

非同期にすると利用者への応答は速くなりますが、状態の管理と再試行の設計が新しく必要になります。次の章から、その中身を実物で見ていきます。

Python と SQLite で作る最小のジョブキュー

ジョブキューの仕組みは、テーブル1つと関数4つで表せます。追加のインストールは不要で、Python の標準ライブラリだけで動きます。

ジョブの状態は次のように移り変わります。

stateDiagram-v2
    [*] --> queued: enqueue
    queued --> running: claim
    running --> done: complete
    running --> queued: release（回数が残っている）
    running --> dead: release（回数を使い切った）
    running --> running: タイムアウト後に別のワーカーが claim
    done --> [*]
    dead --> [*]

miniqueue.py として保存します。

"""SQLite で動く最小のジョブキュー（Python 標準ライブラリのみ）"""
import json
import sqlite3
import time

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    name          TEXT    NOT NULL,              -- ジョブ名（どの処理を呼ぶか）
    payload       TEXT    NOT NULL,              -- 引数（JSON）
    status        TEXT    NOT NULL DEFAULT 'queued',  -- queued / running / done / dead
    attempts      INTEGER NOT NULL DEFAULT 0,    -- 取り出した回数
    max_attempts  INTEGER NOT NULL DEFAULT 3,    -- 最大試行回数
    run_after     REAL    NOT NULL DEFAULT 0,    -- この時刻以降に取り出してよい
    locked_until  REAL,                          -- この時刻まで他のワーカーは触らない
    last_error    TEXT
);
"""


def connect(path):
    # isolation_level=None で、トランザクションを BEGIN / COMMIT で明示する
    conn = sqlite3.connect(path, timeout=5, isolation_level=None)
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA)
    return conn


def enqueue(conn, name, payload, max_attempts=3):
    cur = conn.execute(
        "INSERT INTO jobs (name, payload, max_attempts) VALUES (?, ?, ?)",
        (name, json.dumps(payload, ensure_ascii=False), max_attempts),
    )
    return cur.lastrowid


def claim(conn, timeout_sec, now=None):
    """実行できるジョブを1件だけ取り出す。取り出しと印付けを1文で行う。"""
    now = time.time() if now is None else now
    conn.execute("BEGIN IMMEDIATE")          # 先に書き込みロックを取る
    row = conn.execute(
        """
        UPDATE jobs
           SET status = 'running', attempts = attempts + 1, locked_until = ?
         WHERE id = (
               SELECT id FROM jobs
                WHERE (status = 'queued'  AND run_after    <= ?)
                   OR (status = 'running' AND locked_until <= ?)  -- 時間切れの回収
                ORDER BY id LIMIT 1)
        RETURNING id, name, payload, attempts, max_attempts
        """,
        (now + timeout_sec, now, now),
    ).fetchone()
    conn.execute("COMMIT")
    return row


def complete(conn, job_id):
    conn.execute(
        "UPDATE jobs SET status = 'done', locked_until = NULL WHERE id = ?", (job_id,)
    )


def release(conn, job, error, wait_sec, now=None):
    """エラーになったジョブを戻す。回数を使い切ったら dead（退避）にする。"""
    now = time.time() if now is None else now
    if job["attempts"] >= job["max_attempts"]:
        conn.execute(
            "UPDATE jobs SET status = 'dead', locked_until = NULL, last_error = ? WHERE id = ?",
            (error, job["id"]),
        )
        return "dead"
    conn.execute(
        "UPDATE jobs SET status = 'queued', locked_until = NULL, run_after = ?, last_error = ? WHERE id = ?",
        (now + wait_sec, error, job["id"]),
    )
    return "queued"


def work_once(conn, handlers, timeout_sec=30, wait_sec=0.2):
    """ジョブを1件処理する。処理したら True、キューが空なら False。"""
    job = claim(conn, timeout_sec)
    if job is None:
        return False
    label = f"job#{job['id']} {job['name']} {job['attempts']}/{job['max_attempts']}回目"
    try:
        handlers[job["name"]](json.loads(job["payload"]))
    except Exception as e:  # 処理の中で起きたエラーはすべてここで受ける
        result = release(conn, job, repr(e), wait_sec)
        print(f"{label}: エラー {e!r} -> {result}")
    else:
        complete(conn, job["id"])
        print(f"{label}: 完了")
    return True

設計上の要点は3つです。

• 取り出しと印付けを1文で行う（UPDATE ... RETURNING）。2文に分けるとどうなるかは後の章で確かめます。
• locked_until（タイムアウト）を持つ。ワーカーが途中で止まっても、時間が過ぎれば別のワーカーが回収できます。
• max_attempts を使い切ったら dead にする。うまくいかないジョブがキューの先頭に居座り続けるのを防ぎます。

RETURNING 句は SQLite 3.35.0 以降で使えます（SQLite 公式ドキュメント）。手元の版は次のコマンドで確認できます。

python3 -c "import sqlite3; print(sqlite3.sqlite_version)"

3.45.1

実行結果で見る：再試行と退避

1つ目のデモでは、2種類のジョブを登録します。メール送信は最初の2回だけ相手が応答しない想定、画像変換は入力が画像ではないため何度実行しても同じエラーになる想定です。

"""デモ1：一時的なエラーは再試行で通り、直らないエラーは dead に退避される"""
import os
import time
import miniqueue as q

DB = "demo1.db"
if os.path.exists(DB):
    os.remove(DB)
conn = q.connect(DB)

calls = {"n": 0}


def send_mail(payload):
    calls["n"] += 1
    if calls["n"] <= 2:                       # 最初の2回は相手が応答しない想定
        raise TimeoutError("メールサーバーが応答しません")
    print(f"  メール送信: {payload['to']}")


def resize_image(payload):
    raise ValueError(f"画像ではありません: {payload['file']}")   # 何度やっても同じ


handlers = {"send_mail": send_mail, "resize_image": resize_image}
q.enqueue(conn, "send_mail", {"to": "taro@example.com"}, max_attempts=5)
q.enqueue(conn, "resize_image", {"file": "memo.txt"}, max_attempts=3)

deadline = time.time() + 5
while time.time() < deadline:
    if not q.work_once(conn, handlers, wait_sec=0.2):
        left = conn.execute("SELECT count(*) FROM jobs WHERE status = 'queued'").fetchone()[0]
        if left == 0:
            break
        time.sleep(0.05)                      # 待ち時間が明けるまで少し待つ

print("--- jobs テーブル ---")
for r in conn.execute("SELECT id, name, status, attempts, max_attempts, last_error FROM jobs"):
    print(tuple(r))

python3 demo1_retry.py

job#1 send_mail 1/5回目: エラー TimeoutError('メールサーバーが応答しません') -> queued
job#2 resize_image 1/3回目: エラー ValueError('画像ではありません: memo.txt') -> queued
job#1 send_mail 2/5回目: エラー TimeoutError('メールサーバーが応答しません') -> queued
job#2 resize_image 2/3回目: エラー ValueError('画像ではありません: memo.txt') -> queued
  メール送信: taro@example.com
job#1 send_mail 3/5回目: 完了
job#2 resize_image 3/3回目: エラー ValueError('画像ではありません: memo.txt') -> dead
--- jobs テーブル ---
(1, 'send_mail', 'done', 3, 5, "TimeoutError('メールサーバーが応答しません')")
(2, 'resize_image', 'dead', 3, 3, "ValueError('画像ではありません: memo.txt')")

• 一時的なエラー（相手が応答しない）は、3回目で通りました。再試行が役に立つ場面です。
• 何度やっても同じエラー（入力が正しくない）は、3回とも同じ結果になり、dead に退避されました。この種類のエラーは再試行しても結果が変わらないため、エラーの種類を見て1回目で退避する作りにすると、無駄な実行を減らせます。

このデモでは待ち時間を固定の 0.2 秒にしています。待ち時間を回を追って延ばす計算式（指数バックオフとジッター）は別のテーマとして、この記事では「定義書に待ち時間の方式を書く欄を設ける」ところまでを扱います。

実行結果で見る：同じジョブが2回動く

非同期処理で必ず押さえておきたいのが、処理は終わったのに、完了の記録ができなかった場合です。ワーカーが処理の直後に止まると、キューから見ればそのジョブは「処理中のまま応答が無い」状態なので、タイムアウト後にもう一度取り出されます。

2つ目のデモは、会員にポイントを 100 加算するジョブでこの状況を再現します。

"""デモ2：処理は終わったのに完了を記録できず、同じジョブがもう一度動く"""
import os
import sys
import json
import sqlite3
import miniqueue as q

use_key = len(sys.argv) > 1 and sys.argv[1] == "--idempotent"
DB = "demo2.db"
if os.path.exists(DB):
    os.remove(DB)
conn = q.connect(DB)
conn.executescript("""
CREATE TABLE points (member_id INTEGER PRIMARY KEY, balance INTEGER NOT NULL);
INSERT INTO points VALUES (1, 0);
CREATE TABLE processed (idempotency_key TEXT PRIMARY KEY);  -- 処理済みの冪等キー
""")


def add_points(payload):
    """ポイント加算。use_key のときは冪等キーの記録と加算を同じトランザクションで行う。"""
    conn.execute("BEGIN IMMEDIATE")
    try:
        if use_key:
            conn.execute("INSERT INTO processed VALUES (?)", (payload["key"],))
        conn.execute(
            "UPDATE points SET balance = balance + ? WHERE member_id = ?",
            (payload["amount"], payload["member_id"]),
        )
        conn.execute("COMMIT")
        print("  加算しました")
    except sqlite3.IntegrityError as e:
        conn.execute("ROLLBACK")
        print(f"  処理済みなので何もしません（{e}）")


q.enqueue(conn, "add_points", {"member_id": 1, "amount": 100, "key": "order-1001-points"})

# ワーカーA：取り出して処理まで終えたが、完了を記録する前にプロセスが止まった想定
now = 1000.0
job = q.claim(conn, timeout_sec=30, now=now)
print(f"ワーカーA: job#{job['id']} を取り出し（{job['attempts']}回目）")
add_points(json.loads(job["payload"]))
print("ワーカーA: complete() を呼ぶ前に停止")

# 30秒のタイムアウトが過ぎると、別のワーカーが同じジョブを回収する
job = q.claim(conn, timeout_sec=30, now=now + 31)
print(f"ワーカーB: job#{job['id']} を取り出し（{job['attempts']}回目）")
add_points(json.loads(job["payload"]))
q.complete(conn, job["id"])

balance = conn.execute("SELECT balance FROM points WHERE member_id = 1").fetchone()[0]
print(f"ポイント残高: {balance}（正しくは 100）")

python3 demo2_duplicate.py

ワーカーA: job#1 を取り出し（1回目）
  加算しました
ワーカーA: complete() を呼ぶ前に停止
ワーカーB: job#1 を取り出し（2回目）
  加算しました
ポイント残高: 200（正しくは 100）

同じジョブが2回動き、ポイントが 200 になりました。 キューの側でこれを完全に防ぐことはできないため、処理の側で「2回目は何もしない」ようにします。それが冪等キーです。同じスクリプトに --idempotent を付けて実行します。

python3 demo2_duplicate.py --idempotent

ワーカーA: job#1 を取り出し（1回目）
  加算しました
ワーカーA: complete() を呼ぶ前に停止
ワーカーB: job#1 を取り出し（2回目）
  処理済みなので何もしません（UNIQUE constraint failed: processed.idempotency_key）
ポイント残高: 100（正しくは 100）

冪等キー（ここでは order-1001-points）を主キーにした processed テーブルへの INSERT と、ポイントの加算を同じトランザクションで行っています。2回目は主キーの重複で INSERT がエラーになり、加算ごと取り消されます。

冪等キーの決め方は次の表を参考にしてください。

ジョブの ID（jobs.id）をそのまま冪等キーにすると、同じ依頼が2回登録されたときに別のキーになってしまいます。「同じ出来事なら同じ値になるもの」 を選びます。

処理そのものを冪等にする書き方

冪等キーを使わなくても、処理の書き方で同じ効果が得られる場合があります。

実行結果で見る：2つのワーカーが同じジョブを取る

ワーカーを2つ以上動かすときは、取り出しの書き方に注意が要ります。「SELECT で1件選び、UPDATE で印を付ける」という2文に分けると、その間に別のワーカーが同じ行を選べてしまいます。

"""デモ3：「SELECT してから UPDATE」の2文に分けると、2つのワーカーが同じジョブを取る"""
import os
import miniqueue as q

DB = "demo3.db"
if os.path.exists(DB):
    os.remove(DB)
a = q.connect(DB)
b = q.connect(DB)
q.enqueue(a, "send_mail", {"to": "taro@example.com"})
q.enqueue(a, "send_mail", {"to": "hanako@example.com"})

PICK = "SELECT id FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 1"
MARK = "UPDATE jobs SET status = 'running', attempts = attempts + 1 WHERE id = ?"

# 2文に分けた取り出し：A と B が同じタイミングで SELECT する
id_a = a.execute(PICK).fetchone()["id"]
id_b = b.execute(PICK).fetchone()["id"]
a.execute(MARK, (id_a,))
b.execute(MARK, (id_b,))
print(f"2文に分けた場合: A は job#{id_a}、B は job#{id_b} を取り出した")
print("  attempts =", a.execute("SELECT attempts FROM jobs WHERE id = ?", (id_a,)).fetchone()[0])

# 1文にまとめた取り出し（miniqueue.claim）
a.execute("UPDATE jobs SET status = 'queued', attempts = 0")
job_a = q.claim(a, timeout_sec=30)
job_b = q.claim(b, timeout_sec=30)
print(f"1文にまとめた場合: A は job#{job_a['id']}、B は job#{job_b['id']} を取り出した")

python3 demo3_race.py

2文に分けた場合: A は job#1、B は job#1 を取り出した
  attempts = 2
1文にまとめた場合: A は job#1、B は job#2 を取り出した

2文に分けた場合は A も B も job#1 を取り出し、試行回数が 2 になりました。miniqueue.claim のように書き込みロックを先に取り、取り出しと印付けを1文で行うと、A と B は別々のジョブを受け取ります。

SQLite の BEGIN IMMEDIATE は、書き込みの文を待たずにトランザクションの開始時点で書き込みロックを取ります。他の接続が書き込み中のときは SQLITE_BUSY になることがあるため（SQLite 公式ドキュメント）、接続時に timeout=5 を指定して、空くまで最大5秒待つようにしています。

PostgreSQL では FOR UPDATE SKIP LOCKED を使う

PostgreSQL には、行単位でロックを取り、他のトランザクションがロック中の行を読み飛ばす SKIP LOCKED があります。公式ドキュメントは、キューのようなテーブルに複数の取り出し側がアクセスするときのロック競合を避ける用途に使えると説明しています。

2つの接続で、オプションごとの違いを確かめます。

"""PostgreSQL 版：FOR UPDATE SKIP LOCKED で、2つのワーカーが別々のジョブを取る"""
import psycopg

DSN = "dbname=appdb"
CLAIM = """
SELECT id, name FROM jobs
 WHERE status = 'queued'
 ORDER BY id
 LIMIT 1
 FOR UPDATE {option}
"""

with psycopg.connect(DSN, autocommit=True) as setup:
    setup.execute("DROP TABLE IF EXISTS jobs")
    setup.execute("""
        CREATE TABLE jobs (
            id     bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
            name   text NOT NULL,
            status text NOT NULL DEFAULT 'queued')
    """)
    setup.execute("INSERT INTO jobs (name) VALUES ('send_mail'), ('resize_image'), ('export_csv')")

a = psycopg.connect(DSN)   # ワーカーA（トランザクションを開いたままにする）
b = psycopg.connect(DSN)   # ワーカーB

row_a = a.execute(CLAIM.format(option="SKIP LOCKED")).fetchone()
print("A (SKIP LOCKED):", row_a)

row_b = b.execute(CLAIM.format(option="SKIP LOCKED")).fetchone()
print("B (SKIP LOCKED):", row_b)
b.rollback()

# 比較：SKIP LOCKED を付けないと、B は A のコミットを待ち続ける。
# 待ちを 1 秒で打ち切る設定にして、エラー文を確認する
b.execute("SET lock_timeout = '1s'")
try:
    b.execute(CLAIM.format(option=""))
except psycopg.errors.LockNotAvailable as e:
    print("B (オプションなし):", type(e).__name__, "-", str(e).splitlines()[0])
b.rollback()

# 比較：NOWAIT は待たずにすぐエラーを返す
try:
    b.execute(CLAIM.format(option="NOWAIT"))
except psycopg.errors.LockNotAvailable as e:
    print("B (NOWAIT):", type(e).__name__, "-", str(e).splitlines()[0])
b.rollback()

# A が処理を終えて状態を更新し、コミットする
a.execute("UPDATE jobs SET status = 'done' WHERE id = %s", (row_a[0],))
a.commit()
print("A がコミットしたあと:", b.execute("SELECT id, status FROM jobs ORDER BY id").fetchall())
a.close()
b.close()

A (SKIP LOCKED): (1, 'send_mail')
B (SKIP LOCKED): (2, 'resize_image')
B (オプションなし): LockNotAvailable - canceling statement due to lock timeout
B (NOWAIT): LockNotAvailable - could not obtain lock on row in relation "jobs"
A がコミットしたあと: [(1, 'done'), (2, 'queued'), (3, 'queued')]

出力中のデータベース名は appdb に置き換えています。

同じドキュメントには、読み飛ばしによって得られるデータの見え方は一貫したものにならないため、一般的な用途には向かないとも書かれています。キューからの取り出しに限って使うのが安全です。

リトライと冪等性の違い、キューとイベント通知の違い

混同されやすい2組を表で整理します。

リトライと冪等性

キューとイベント通知

どれを使うかの判断表

ジョブ定義書テンプレート

ジョブを1つ追加するたびに、次の表を1枚埋めます。コードを書く前に埋めておくと、再試行と退避の方針がレビューで確認できます。

記入例：備品予約システムの予約確定メール

架空の備品予約システムを題材にした記入例です。数値はすべて例です。

表計算ソフトに貼れるタブ区切り版です。

項目  記入内容    記入例
ジョブ名    コード上の名前と日本語名    send_reservation_mail（予約確定メール送信）
起点  どの操作・どのデータの発生で登録されるか    利用者が予約を確定したとき
引数  ジョブに渡す値 予約ID
最大試行回数  初回を含めて何回まで実行するか 5回
待ち時間    再試行までの間隔の方式と上限  指数バックオフ（初回10秒、上限10分）＋ジッター
タイムアウト  1回の実行に許す時間  30秒
冪等キー    何を組み合わせて作るか 予約ID＋confirm-mail
再試行するエラー    一時的とみなすエラーの種類   接続エラー、応答の時間切れ、一時的な混雑を示す応答
再試行しないエラー   すぐ退避するエラーの種類    宛先アドレスの形式エラー、予約が取り消し済み
エラー時の退避先    回数を使い切ったジョブをどこに残すか  jobsテーブルのstatusをdeadにして残す
通知先 退避されたときに誰へ何で知らせるか   運用担当のチャットにジョブ名・予約ID・最後のエラーを投稿
再実行の手順  退避したジョブをどう戻すか   statusをqueued、attemptsを0に戻す
保存期間    完了・退避したジョブをいつまで残すか  doneは30日、deadは対応が終わるまで

数値を決めるときの考え方

確かめて分かったこと

今回の実行で分かった、見落としやすい点です。

1. 取り出しを2文に分けると、2つのワーカーが同じ job#1 を取りました。 エラーは出ず、試行回数が 2 になるだけなので、気づきにくい動きです。
2. 完了の記録の前にワーカーが止まると、タイムアウト後に同じジョブがもう一度動き、ポイントは 200 になりました。 冪等キーを加算と同じトランザクションに入れると 100 のままでした。
3. 入力に原因があるエラーも、指定しなければ最大回数まで再試行されました。 デモでは同じ ValueError が3回出てから退避されています。エラーの種類で分ける欄を定義書に設けた理由です。
4. 完了したジョブの last_error に、前回までのエラーが残りました。 job#1 は done ですが、last_error は TimeoutError のままです。経緯として残すのか、完了時に消すのかを決めておくと、一覧を見たときに迷いません。
5. PostgreSQL でオプションなしの FOR UPDATE は、相手のコミットを待ちました。 lock_timeout を 1 秒にすると「canceling statement due to lock timeout」、NOWAIT では「could not obtain lock on row in relation "jobs"」というエラー文になり、SKIP LOCKED だけが次のジョブを返しました。

よくある質問

Q. 同期処理と非同期処理の違いは？ A. 結果を応答の中で返すのが同期、受付だけを返して本体を後から行うのが非同期です。非同期では、状態の管理・再試行・冪等性の設計が加わります。

Q. 非同期処理にする判断基準は？ A. 「結果をその場で必要としない」「数秒を超える」「外部への呼び出しを含む」のどれかに当てはまれば非同期にします。起点が時刻でまとめて処理できるものは定期バッチにします。

Q. ジョブキューのリトライ回数は何回にすればよいですか？ A. 待ち時間の合計が、利用者に待ってもらえる時間に収まる回数にします。回数だけを先に決めるのではなく、待ち時間の方式と合わせて定義書に書きます。

Q. 同じジョブが二重実行されるのを防げないときは？ A. 処理の側を冪等にします。冪等キーを一意制約つきのテーブルに記録し、本体の更新と同じトランザクションにまとめると、2回目は何も変えずに終わります。

Q. デッドレターキューとは？ A. 最大試行回数を使い切ったジョブを退避しておく場所です。この記事の例では jobs テーブルの status を dead にして残しています。引数と最後のエラーを残し、原因を直したあとに戻せるようにします。

まとめ

• 結果をその場で必要としない処理、数秒を超える処理、外部への呼び出しを含む処理は非同期にします。
• 時刻が起点でまとめて処理できるものは定期バッチ、それ以外の短い処理は同期のままにします。
• ジョブキューはテーブル1つと、登録・取り出し・完了・戻しの4つの関数で表せます。
• 取り出しは1文で行うか、行ロック（PostgreSQL の FOR UPDATE SKIP LOCKED）を使うと、ワーカーを増やしても同じジョブを取り合いません。
• 同じジョブは2回動くことがあるため、リトライと冪等キーをセットで設計します。
• 回数を使い切ったジョブは退避して通知し、原因を直したあとに戻せるようにします。
• ジョブを追加するときは、ジョブ定義書に回数・待ち時間・タイムアウト・冪等キー・退避先・通知先を書きます。

参考・出典

• PostgreSQL 16 Documentation「SELECT」The Locking Clause（NOWAIT と SKIP LOCKED の説明）: https://www.postgresql.org/docs/16/sql-select.html
• SQLite Documentation「Transaction」（DEFERRED／IMMEDIATE／EXCLUSIVE の説明）: https://www.sqlite.org/lang_transaction.html
• SQLite Documentation「RETURNING」（対応する版）: https://www.sqlite.org/lang_returning.html

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。