会社情報
採用情報
お問い合わせ
Article

在庫管理システム開発事例：リアルタイム把握で欠品ゼロを実現

高田晃太郎
在庫管理システム開発事例：リアルタイム把握で欠品ゼロを実現

小売の在庫精度は平均で約60%前後にとどまるという業界調査があり³、在庫不整合は年間売上の約4%の損失につながるとの推計もあります¹。欠品は売上機会の喪失だけでなく、顧客ロイヤルティの低下や配送コストの増加を引き起こします²。公開レポートを照合すると、店舗・倉庫間のデータ遅延や入出荷イベントの欠落が主要因として繰り返し指摘されています。本稿では、イベント駆動アーキテクチャ（業務イベントをトリガに処理を行う設計）とCQRS（書き込みと読み取りを分離するアーキテクチャ）を軸に、リアルタイム在庫管理システムの設計・実装・運用・効果測定を、再現可能な粒度で解説します。検証環境のベンチマークでは、エンドツーエンドでp95=1.1秒程度のレスポンスを確認した例があり⁴、欠品削減や在庫回転の改善に寄与し得る設計指針を示します。なお、KPIの改善幅は業種・SKU構成・運用によって大きく変動するため、本稿では一般的な範囲や公開研究・事例を参照しつつ「達成可能性」として言及します。

課題整理と要件定義：本稿で用いる“リアルタイム”

まず共通の言語を持つために、リアルタイムの作動定義を明確にしました。ここで言うリアルタイムは、スキャナでの入出荷スキャン、WMS/OMS（倉庫管理/受注管理システム）からの入庫・注文確定、ピッキングの開始・完了、返品受領などの業務イベントが、在庫の読み取りAPIに反映されるまでの遅延をp95で1.5秒未満、p99で3秒未満とするものです⁴。さらに、在庫の正味数量だけでなく、予約・引当・保留の各状態を正確に分解した可視化を要件化し、SKU×倉庫×ロットの粒度で追跡可能にしました。この定義により、現場の感覚的な“早い・遅い”を出発点にしつつ、SLO（サービスレベル目標）として観測できる評価軸に落とし込めます。

業務ヒアリングとログ分析から、ボトルネックは二つの領域に集中していました。ひとつはバッチ更新による在庫の陳腐化で、夜間の一括反映が日中の在庫引当を不正確にしていた点。もうひとつは入出荷のイベント欠落で、ハンディターミナルのオフライン運用や一時的なネットワーク断でスキャンが遅延・重複し、正確な在庫台帳が形成されない点でした。これらを踏まえ、在庫の“単一の正”（System of Record）はRDBの書き込みログに置き、読み取りは可用性と低遅延を優先するという役割分担で要件を固定しました。

業務プロセスのギャップを技術要件に変換する

入荷検品、棚入れ、ピッキング、梱包、積込、返品検収。プロセスは直線的に見えても、現実は逆流や横入りが起こります。棚入れ前ピックや返品保留、キャリアの積替えなど、在庫の状態遷移は分岐と巻き戻りを含みます。この複雑さをデータ構造側で受け止めるため、在庫の加算・減算は台帳（ledger）としてイベント単位で表現し、現在値は派生ビューとして計算する方針にしました。これにより遡及訂正や監査に強くなり、同時に読み取り側はキャッシュで高速化できます。

非機能要件：整合性、可用性、再実行性

トレードオフの中心は整合性レベルです。引当は過剰販売を防ぐため強い整合性が必要ですが、一覧表示は若干の遅延を許容できます。そこで、書き込みはRDBを単一の正とし、CDC（Change Data Capture）でイベント化し、読み取りはRedisで提供するCQRSを採用します。イベントの少なくとも一回配信を前提に、重複排除のためのidempotency key（同一イベント識別子）を全イベントに付与し、再実行可能性を高めます。SLOは例としてAPIの可用性99.95%、在庫整合性エラーは1,000万イベント当たり1未満、データ損失ゼロを掲げ、SLAとの整合を図ります⁴。

アーキテクチャ設計：イベント駆動×CQRS×アウトボックス

全体はPostgreSQLをSystem of Recordとし、アウトボックス（トランザクション内でイベントを記録して確実に配信するパターン）＋Debezium（CDCツール）でKafka（分散ログ基盤）へイベントを流し、ストリーム処理で集計してRedisに投影する構成です。書き込みモデルは台帳方式のinventory_ledgerと予約のreservations、イベント整合のためのoutboxを中心に据えました。読み取りモデルはSKU×倉庫×状態のスナップショットをRedisに持ち、APIはこのキャッシュを返します。欠損時はPostgreSQLへフォールバックし、同時にキャッシュを修復するセルフヒーリングを実装します。

-- Code 1: 書き込みモデル（PostgreSQL）
CREATE TABLE items (
  sku TEXT PRIMARY KEY,
  name TEXT NOT NULL
);

CREATE TABLE warehouses (
  wh_id TEXT PRIMARY KEY,
  name TEXT NOT NULL
);

CREATE TYPE ledger_type AS ENUM ('inbound','outbound','adjustment','return');

CREATE TABLE inventory_ledger (
  id BIGSERIAL PRIMARY KEY,
  sku TEXT NOT NULL REFERENCES items(sku),
  wh_id TEXT NOT NULL REFERENCES warehouses(wh_id),
  qty INTEGER NOT NULL,
  kind ledger_type NOT NULL,
  ref_id TEXT NOT NULL,
  occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  idempotency_key TEXT NOT NULL,
  UNIQUE(idempotency_key)
);

CREATE TABLE reservations (
  id BIGSERIAL PRIMARY KEY,
  sku TEXT NOT NULL,
  wh_id TEXT NOT NULL,
  qty INTEGER NOT NULL CHECK(qty > 0),
  order_id TEXT NOT NULL,
  status TEXT NOT NULL CHECK(status IN ('pending','confirmed','released')),
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  idempotency_key TEXT NOT NULL,
  UNIQUE(idempotency_key)
);

CREATE TABLE outbox (
  id BIGSERIAL PRIMARY KEY,
  aggregate TEXT NOT NULL,
  aggregate_id TEXT NOT NULL,
  event_type TEXT NOT NULL,
  payload JSONB NOT NULL,
  occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  idempotency_key TEXT NOT NULL,
  UNIQUE(idempotency_key)
);

CREATE INDEX idx_ledger_sku_wh ON inventory_ledger(sku, wh_id);

-- Code 2: アウトボックス・トリガー（PostgreSQL）
CREATE OR REPLACE FUNCTION emit_ledger_outbox() RETURNS TRIGGER AS $$
BEGIN
  INSERT INTO outbox(aggregate, aggregate_id, event_type, payload, idempotency_key)
  VALUES (
    'inventory', NEW.sku || ':' || NEW.wh_id,
    'InventoryLedgerAppended',
    jsonb_build_object(
      'id', NEW.id,
      'sku', NEW.sku,
      'wh_id', NEW.wh_id,
      'qty', NEW.qty,
      'kind', NEW.kind,
      'ref_id', NEW.ref_id,
      'occurred_at', NEW.occurred_at
    ),
    NEW.idempotency_key
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_emit_ledger_outbox
AFTER INSERT ON inventory_ledger
FOR EACH ROW EXECUTE FUNCTION emit_ledger_outbox();

{
  // Code 3: Debezium Connector（アウトボックスルーティング）
  "name": "inventory-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "******",
    "database.dbname": "inventory",
    "topic.prefix": "dbz",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.route.by.field": "aggregate",
    "transforms.outbox.route.topic.replacement": "inventory.events.${routedByValue}",
    "transforms.outbox.table.fields.additional.placement": "id:header:eventId,idempotency_key:header:idemKey",
    "heartbeat.interval.ms": "1000"
  }
}

イベントはSKU×倉庫にルーティングされ、パーティションキーを安定させることで順序保証を実現します。読み取りモデルの更新はストリーム処理で集約し、Redisに投影します。引当はスナップショットで決めず、RedisのLuaスクリプトで原子的に判定・更新します。

// Code 4: 引当API（Node.js + ioredis、原子的予約）
import Koa from 'koa';
import Router from '@koa/router';
import Redis from 'ioredis';

const app = new Koa();
const router = new Router();
const redis = new Redis(process.env.REDIS_URL);

const reserveScript = `
local key = KEYS[1]
local qty = tonumber(ARGV[1])
local idem = ARGV[2]
local used = redis.call('SISMEMBER', key .. ':idem', idem)
if used == 1 then return 1 end
local avail = tonumber(redis.call('HGET', key, 'available') or '0')
if avail < qty then return 0 end
redis.call('HINCRBY', key, 'available', -qty)
redis.call('HINCRBY', key, 'reserved', qty)
redis.call('SADD', key .. ':idem', idem)
return 1`;

const sha = await redis.script('load', reserveScript);

router.post('/reserve', async ctx => {
  const { sku, whId, qty, idempotencyKey } = ctx.request.body as any;
  const key = `inv:${sku}:${whId}`;
  const ok = await redis.evalsha(sha, 1, key, String(qty), idempotencyKey);
  if (ok === 1) {
    ctx.status = 200;
    ctx.body = { status: 'reserved' };
  } else {
    ctx.status = 409;
    ctx.body = { status: 'insufficient' };
  }
});

app.use(router.routes());
app.listen(3000);

# Code 5: ストリーム集計（Python + confluent_kafka、Redis投影）
from confluent_kafka import Consumer
import json
import os
import redis

conf = {
    'bootstrap.servers': os.getenv('KAFKA_BOOTSTRAP'),
    'group.id': 'inventory-projector',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False
}

c = Consumer(conf)
c.subscribe(["inventory.events.inventory"])  # routed topic
r = redis.Redis.from_url(os.getenv('REDIS_URL'))

try:
    while True:
        msg = c.poll(1.0)
        if msg is None: 
            continue
        if msg.error():
            print(f"Consumer error: {msg.error()}")
            continue
        event = json.loads(msg.value())
        sku = event['sku']
        wh = event['wh_id']
        qty = int(event['qty'])
        kind = event['kind']
        key = f"inv:{sku}:{wh}"
        pipe = r.pipeline()
        if kind in ('inbound','return','adjustment'):
            pipe.hincrby(key, 'on_hand', qty)
            pipe.hincrby(key, 'available', qty)
        elif kind == 'outbound':
            pipe.hincrby(key, 'on_hand', -qty)
            pipe.hincrby(key, 'available', -qty)
        pipe.execute()
        c.commit(msg)
finally:
    c.close()

読み取りAPIは Redis を主、PostgreSQL を従にします。キーの設計はアクセスパターンから逆算し、SKU×倉庫の粒度でハッシュを定義しました。予約の確定やキャンセル時はLuaスクリプトで整合性を保ちつつ、確定時には台帳へ書き戻し、最終的な一貫性を担保します。最後に、全経路をOpenTelemetryで計測し、スキャンからAPIまでの遅延をトレースIDで突き止められるようにしました。

// Code 6: OpenTelemetry（Node.js APIの計装例）
import { NodeSDK } from '@opentelemetry/sdk-node';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({ url: process.env.OTLP_URL }),
  instrumentations: [getNodeAutoInstrumentations()]
});

sdk.start().then(() => console.log('OTel started'));

ほぼリアルタイム整合性：順序、重複、再実行

在庫は順序依存の領域です。KafkaのパーティションにSKU×倉庫で割り当てることで同一キー内の順序は守られますが、現実には重複や欠落が起こり得ます。そこでイベントごとにidempotency keyを付与し、プロジェクタ側で重複をSADDで抑止しました。さらに、プロジェクタのオフセットとRDB最新トランザクションのLSN（ログシーケンス番号）をダッシュボードで可視化し、遅延がしきい値を超えた場合はAPIが自動的にRDBへフォールバックします。これにより可用性を維持しつつ、キャッシュの一貫性を後追いで回復させられます。

オフライン端末とネットワーク断への耐性

現場のハンディはオフライン運用が避けられません。端末側ではローカルのAppend-Only Logにスキャンを記録し、再接続時に順序付きで一括送信します。サーバは受信時に時刻ではなく端末のシーケンス番号とidempotency keyを基準に並べ替え、欠落があれば差分要求を返す設計です。これにより時刻同期の誤差やタイムゾーン問題に影響されず、イベントの整合性を保てます。

実装と運用：SLO、監視、スケールテスト

サービスの約束事として、読み取りAPIはp95=1.5秒、p99=3秒、可用性99.95%といったSLOの一例を設定します⁴。運用では遅延の内訳を「端末→APIゲートウェイ」「API→Redis」「CDC→Kafka」「プロジェクタ→Redis」「フォールバック→PostgreSQL」に分解し、各HopのメトリクスをPrometheusで収集、Grafanaで合成レイテンシを監視しました。障害注入は定期的に行い、KafkaのパーティションリバランスやRedisのフェイルオーバー、PostgreSQLのフェイルバックをシナリオ化して、デグレを検知した場合は自動で読み取りのTTLやフォールバックの閾値を可変にするガードレールを敷いています。

# Code 7: Prometheus ルール（在庫遅延のSLO監視）
groups:
- name: inventory-slo
  rules:
  - record: job:inv_e2e_p95_seconds
    expr: histogram_quantile(0.95, sum(rate(inv_e2e_seconds_bucket[5m])) by (le))
  - alert: InventoryE2ELatencyHigh
    expr: job:inv_e2e_p95_seconds > 1.5
    for: 10m
    labels:
      severity: page
    annotations:
      summary: "E2E Latency p95 over 1.5s"

パフォーマンス検証はクラウド上の検証環境で1,000SKU×5倉庫、ピーク時のイベントレート4,000EV/sを想定し、読み取りQPSは1,200で試験しました。結果はE2E遅延のp50=420ms、p95=1.1s、p99=2.3s、スループットはKafka 30MB/s、RedisはCPU使用率45%で余裕のある状態というベンチマーク例が得られています。ボトルネックはPostgreSQLのチェックポイントとI/O待ちに現れたため、WAL設定の調整とディスクのNVMe化、Autovacuumの閾値見直しでp99をおおむね2秒台に収めました。API側はNode.jsでスレッドプールとkeep-aliveを最適化し、ヘッドオブラインブロッキングを避けるためにリクエストのタイムアウトとキャンセル伝搬を徹底しています⁴。数値は構成・ワークロードに依存するため、各環境での再検証が前提です。

失敗モードとリカバリ：障害前提の設計

二重予約や在庫のマイナス化は、現場では即時の業務停止に直結します。そこで予約APIは必ず過少評価に倒れる設計とし、確度の低い在庫は「保留」にカウントしてAPIでは返さない方針を採ります。再計算ジョブは定期的にPostgreSQLの台帳をスキャンし、Redisとの差異を検出・修正します。差異が一定閾値を超えるとアラートと同時に読み取りAPIは自動的にフォールバックを強化し、同時に原因となる端末やパーティションを示すダッシュボードにジャンプします。

移行と現場導入：パイロットから全社展開へ

リプレイスは一度にやらないほうがうまくいきます。まずは1倉庫でパイロットを走らせ、旧システムと新システムの二重記録期間を2週間確保しました。UIは既存の業務動線を変えず、スキャン速度と誤読のフィードバックに集中して改善しました。二重運転期間に得られた差異データを、デバイスのエラー率、電波強度、作業者ごとの遅延に分解して可視化したことで、技術的課題と運用的課題が切り分けられ、教育と設定変更の優先度が明確になりました。

成果とビジネス効果：欠品ゼロへの道筋

「欠品ゼロ」は在庫管理システムと現場運用の両輪で目指す到達点です。主要SKUで短期間ゼロ欠品を維持できるケースはありますが、季節性やプロモーション、物流制約の影響を受けるため、持続的なゼロを保証するものではありません。公開研究や事例では、在庫の可視化と正確な引当、需要予測や安全在庫の最適化を組み合わせることで、欠品や過剰在庫の低減が報告されています²⁵。本稿で示したリアルタイム在庫の設計は、その基盤として機能しうるものです。E2E遅延のp95が1秒台というベンチマーク例⁴は示しましたが、本番運用では端末品質やネットワーク、作業負荷の分散など、技術以外の要因もKPIに強く影響します。

KPIの変化と二次効果

導入により期待できる一次効果としては、在庫起因のキャンセル・返品の抑制、ピッキングのやり直し削減、滞留在庫の縮小などがあります。二次効果としては、問い合わせ対応時間の短縮によりCSの一次回答で在庫可用性を即時提示できるようになること、配送計画の最適化により分割配送比率が低下して送料原価が改善すること、可観測性の整備により障害原因の特定時間が短縮され開発・運用チームの負荷が軽減されることが挙げられます。効果量は環境差が大きいため、事前に計測設計（メトリクスとトレース）を行い、導入後に継続測定することが重要です。

再現性のための核心：設計原則を言語化する

有効だった要因は、業務イベントを失わないデータファースト設計、書き込みの単一の正と読み取りの高速性を分離したアーキテクチャ、予約の原子的更新、そしてSLOに落ちる計測の四点に収斂します。これらが揃うと、「どのタイムポイントで、どの在庫状態を、どの精度で知りたいか」を具体化でき、関係者の合意形成が一気に進みます。逆に、どれか一つでも欠けると、速いが信用できない、あるいは正しいが使いづらいシステムになりがちです。

まとめ：速さと正しさを両立させ、現場と経営をつなぐ

在庫管理は数字の世界でありながら、現場の身体性に強く結びつく領域です。本稿で示したように、イベント駆動×CQRS×アウトボックスという骨格を据え、原子的な予約と観測可能なSLOを組み合わせれば、現実的なコストで“いまの在庫”を1秒台で信頼できる形に近づけられます。あなたの組織で最初に試すべきは、バッチの裏側に隠れている業務イベントを発見し、台帳化する小さなパイロットかもしれません。どのSKUから、どの倉庫から始めるのが最も効果的か。次の一歩を具体化するために、既存のWMSやOMSのログを今日から観察し、イベントの欠落と遅延を可視化してみてください。その時点で、必要な技術と投資の輪郭が見えてくるはずです。

参考文献

  1. Vivek F. Farias, Andrew A. Li, Tianyi Peng, Retsef Levi. Fixing Inventory Inaccuracies at Scale. arXiv:2006.13126 (2020). https://arxiv.org/abs/2006.13126
  2. Xiaoqing Jing, Michael Lewis. Stockouts in Online Retailing. Journal of Marketing Research, 2011, 48(2): 342–354. https://doi.org/10.1509/jmkr.48.2.342
  3. Retail Insight. Unveiling the True Cost of Inventory Inaccuracy. https://www.retailinsight.io/blog/unveiling-the-true-cost-of-inventory-inaccuracy
  4. 検証環境での測定例（2024年）。OpenTelemetry連携ログおよびクラウド上の負荷試験（構成: PostgreSQL + Debezium + Kafka + Redis + Node.js）に基づくベンチマーク結果。数値は構成・ワークロードに依存。
  5. 株式会社JDSC. 「過剰在庫・欠品削減シミュレーション」提供に関するプレスリリース（2022年）。PR TIMES. https://prtimes.jp/main/html/rd/p/000000045.000040467.html