Kafka入門｜KRaftでトピック作成から消費まで最短手順

Apache Kafkaとは、大量のメッセージ（イベント）を順番どおりにディスクへ記録し、複数のアプリが好きなタイミングで読み出せる「分散ログ」型のメッセージ基盤です。 送る側（プロデューサー）と受け取る側（コンシューマー）を切り離せるので、注文・在庫・通知のようにシステムをまたいでデータを流す仕組みに使われます。

Kafkaは4.0でZooKeeperが取り除かれ、KRaft（Kafka自身でメタデータを管理する方式）だけになりました。この記事では、2026年10月時点の最新版 Kafka 4.3.1 を前提に、フォーマット → 起動 → トピック作成 → 送信 → 受信 → グループ確認の最短手順を公式Quickstartに沿って紹介します。あわせて、用語の図解、Pythonのコピペ用コード、パーティション数・acks・retentionの決め方表、配信保証の比較、RabbitMQなどとの使い分けまで、入門者が最初につまずく所をまとめました。

検証について：この記事の作成環境ではKafka本体（Apache公式のダウンロード、Dockerイメージ）を入手できなかったため、Kafkaのコマンドは公式ドキュメントに沿って記載し、未実行です。Pythonコードは confluent-kafka 2.15.1（Python 3.11）で構文・importを確認し、ブローカーがない状態で実行したときの実際のログを載せています。

Kafka入門の早見表

Kafkaの用語を図で理解する

Kafkaは「トピック（名前のついたログ）」を「パーティション（分割された追記専用のファイル）」に分けて、複数のブローカー（サーバー）に置きます。 プロデューサーが末尾に追記し、コンシューマーは「オフセット（何番目まで読んだか）」を記録しながら読み進めます。

flowchart LR
    P1[プロデューサー<br/>注文サービス] -->|key=A001| T0
    P1 -->|key=A002| T1
    subgraph TP["トピック orders"]
        T0[パーティション0<br/>offset 0,1,2...]
        T1[パーティション1<br/>offset 0,1,2...]
        T2[パーティション2<br/>offset 0,1,2...]
    end
    subgraph G1["コンシューマーグループ order-service"]
        C1[コンシューマー1]
        C2[コンシューマー2]
    end
    T0 --> C1
    T1 --> C1
    T2 --> C2
    subgraph G2["別のグループ mail-service"]
        C3[コンシューマー3]
    end
    T0 -. 同じデータを独立して読む .-> C3
    T1 -.-> C3
    T2 -.-> C3

「読んでも消えない」「グループごとに独立して読める」がキュー型の仕組みとの一番の違いです。同じ注文イベントを、在庫サービスとメールサービスがそれぞれ自分のペースで読めます。

Kafka 4.xで変わったこと

古い記事にある --zookeeper localhost:2181 オプションや zookeeper-server-start.sh は4.xでは使いません。 すべてのコマンドは --bootstrap-server でブローカーに直接つなぎます。

KRaftで動かす最短手順（公式Quickstartに沿って）

手順は「ダウンロード → クラスターIDを作る → フォーマット → 起動 → トピック作成 → 送受信」の6ステップです。ここではブローカーとコントローラーを1台で兼ねる、学習用の構成で動かします（未実行。公式Quickstartのコマンドに沿って記載）。

1. ダウンロードして展開する

公式サイトのダウンロードページから kafka_2.13-4.3.1.tgz を取得します（2.13 は内部で使うScalaの版で、利用者は気にしなくて大丈夫です）。

java -version          # 17以上であることを確認
tar -xzf kafka_2.13-4.3.1.tgz
cd kafka_2.13-4.3.1

2. クラスターIDを作ってストレージをフォーマットする

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties

KRaftでは、起動の前にログディレクトリをフォーマットする必要があります。 --standalone は「コントローラー1台だけのクォーラム（合議の単位）を作る」という指定です。複数台のコントローラーで組む場合は --initial-controllers で参加者を列挙します。

config/server.properties の主な項目は次のとおりです（KRaftの1台構成）。

process.roles=broker,controller
node.id=1
controller.quorum.bootstrap.servers=localhost:9093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
controller.listener.names=CONTROLLER
log.dirs=/tmp/kraft-combined-logs
num.partitions=1

3. ブローカーを起動する

bin/kafka-server-start.sh config/server.properties

このターミナルは起動したままにして、以降は別のターミナルで操作します。

4. トピックを作る

bin/kafka-topics.sh --create --topic orders \
  --partitions 3 --replication-factor 1 \
  --bootstrap-server localhost:9092

bin/kafka-topics.sh --describe --topic orders --bootstrap-server localhost:9092

--describe では、パーティションごとに Leader（リーダーのノード番号）、Replicas（レプリカの置き場所）、Isr（追いついているレプリカ）が表示されます。1台構成では --replication-factor は1までです。

5. メッセージを送る（console-producer）

bin/kafka-console-producer.sh --topic orders --bootstrap-server localhost:9092
>注文A001を受け付けました
>注文A002を受け付けました

1行が1レコードになります。Ctrl-C で終了します。キーを付けて送りたいときは次のようにします。

bin/kafka-console-producer.sh --topic orders --bootstrap-server localhost:9092 \
  --property parse.key=true --property key.separator=:
>A001:ノート2冊
>A002:ペン5本

6. メッセージを受け取る（console-consumer）

bin/kafka-console-consumer.sh --topic orders --from-beginning \
  --bootstrap-server localhost:9092 \
  --property print.key=true --property print.partition=true

--from-beginning を付けないと「今から届くもの」だけが表示されます。何度実行しても同じデータが読めるのは、読んでもデータが消えないためです。

（コマンドラインの --property は新しい版で --reader-property／--formatter-property などへ名前の整理が進んでいます。手元の版の --help で確認してください。）

7. コンシューマーグループを確認する

# グループ名を付けて受信（別ターミナルで2つ起動すると、パーティションを分担する）
bin/kafka-console-consumer.sh --topic orders --group order-service \
  --bootstrap-server localhost:9092

# グループの一覧と状態
bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
bin/kafka-consumer-groups.sh --describe --group order-service --bootstrap-server localhost:9092

--describe の主な列は次のとおりです。

オフセットの巻き戻しは、グループのコンシューマーをすべて止めてから行います。--dry-run で確認してから --execute にすると安心です。

bin/kafka-consumer-groups.sh --reset-offsets --to-earliest \
  --group order-service --topic orders --dry-run --bootstrap-server localhost:9092

Dockerで試す場合

docker run -p 9092:9092 apache/kafka:4.3.1

公式イメージは既定でKRaftの1台構成が起動します。コマンドはコンテナの中の /opt/kafka/bin/ にあります（例：docker exec -it <コンテナ名> /opt/kafka/bin/kafka-topics.sh ...）。

Pythonでプロデューサーとコンシューマーを作る（confluent-kafka）

PythonからはConfluentが提供する confluent-kafka（C言語のlibrdkafkaを使う公式系クライアント）がよく使われます。pip install confluent-kafka だけで入り、ライブラリのビルドは不要です。

プロデューサー（producer.py）

# pip install confluent-kafka   （確認：confluent-kafka 2.15.1 / Python 3.11）
import json
from confluent_kafka import Producer

conf = {
    "bootstrap.servers": "localhost:9092",
    "acks": "all",                 # 全ISRへの書き込みを待つ
    "enable.idempotence": True,    # 再送しても重複しない
    "linger.ms": 5,                # 少し待ってまとめて送る
}
producer = Producer(conf)

def on_delivery(err, msg):
    if err is not None:
        print(f"送信失敗: {err}")
    else:
        print(f"送信OK: topic={msg.topic()} partition={msg.partition()} offset={msg.offset()} key={msg.key().decode()}")

orders = [
    {"order_id": "A001", "item": "ノート", "qty": 2},
    {"order_id": "A002", "item": "ペン", "qty": 5},
    {"order_id": "A001", "item": "ノート", "qty": 1},
]
for o in orders:
    producer.produce(
        "orders",
        key=o["order_id"],                       # 同じキーは同じパーティションへ
        value=json.dumps(o, ensure_ascii=False),
        on_delivery=on_delivery,
    )
    producer.poll(0)                             # コールバックを処理

remaining = producer.flush(10)                   # 最大10秒、送信完了を待つ
print("未送信:", remaining)

produce() は送信を予約するだけで、実際の送信はバックグラウンドで行われます。 終了前に flush() を呼ばないと、送信待ちのメッセージが残ったままプログラムが終わります。

ブローカーを起動していない状態で実行したときの実際の出力です（Kafkaに届かない場合の動きの確認）。

%3|1790860397.125|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT)
%3|1790860397.177|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT, 1 identical error(s) suppressed)
未送信: 3

接続できなくても produce() 自体はエラーにならず、flush() の戻り値（未送信件数）に表れます。 送信の成否は on_delivery のコールバックか flush() の戻り値で確認します。ブローカーが動いていれば、コールバックで 送信OK: topic=orders partition=... offset=... が表示されます（この行はブローカーなしでは確認できていません）。

コンシューマー（consumer.py）

# pip install confluent-kafka
import json
from confluent_kafka import Consumer, KafkaError, KafkaException

conf = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "order-service",       # 同じgroup.idのコンシューマーでパーティションを分担
    "auto.offset.reset": "earliest",   # コミット済みオフセットがなければ先頭から
    "enable.auto.commit": False,       # 処理が終わってから手動でコミット
}
consumer = Consumer(conf)
consumer.subscribe(["orders"])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise KafkaException(msg.error())
        order = json.loads(msg.value())
        print(f"受信: partition={msg.partition()} offset={msg.offset()} {order}")
        # ここで注文を処理する
        consumer.commit(message=msg, asynchronous=False)
except KeyboardInterrupt:
    pass
finally:
    consumer.close()   # グループから抜けて再割り当てを早める

「処理してからコミット」にすると、途中で止まっても未処理のメッセージを読み直せます（at-least-once）。 そのかわり同じメッセージを2回処理することがあるので、処理側は「同じ注文IDなら1回だけ反映する」ように作っておきます。

設定の決め方：パーティション数・acks・retention・レプリカ数

acks の比較

配信保証の比較

exactly-onceはKafkaの中で完結する処理（読んで加工して別トピックへ書く）が対象です。外部のデータベースへ書く場合は、注文IDで重複を判定するなど処理側の工夫と組み合わせます。

RabbitMQなどとの使い分け

「同じイベントを複数のシステムが使う」「あとから読み直したい」ならKafka、「タスクを1回ずつ確実に誰かへ渡したい」ならRabbitMQやクラウドのキューが選びやすいです。Kafka 4.2以降は共有グループ（Queues for Kafka）で、キューに近い使い方もできるようになりました。

よくあるエラーと対処

Kafka本体は未実行のため、下表のメッセージは公式ドキュメント・クライアントの例外名に基づきます（1行目のみ、この記事の環境で実際に出力を確認）。

よくある質問

Q. ZooKeeperはもう不要ですか？ A. はい。Kafka 4.0以降はKRaftのみで、ZooKeeperは使いません。 3.x以前のZooKeeper構成から上げる場合は、3.x系の途中で「ZooKeeperからKRaftへの移行」を済ませてから4.xへ上げる流れが公式に案内されています。

Q. パーティション数はいくつにすればよいですか？ A. 同時に処理したいコンシューマーの数以上が目安です。増やすことはできても減らせないこと、増やすとキーの割り当て先が変わることから、将来の台数を見込んで少し多めにします。学習用なら3で十分です。

Q. メッセージの順序は保証されますか？ A. 同じパーティションの中だけ保証されます。 注文ごとに順序が必要なら、注文IDをキーにして同じパーティションに入れます。トピック全体で厳密な順序が必要なら、パーティションを1つにする方法もありますが、並列処理はできなくなります。

Q. 読んだメッセージはいつ消えますか？ A. 読んだかどうかとは関係なく、retention.ms（既定7日）や retention.bytes を超えた古いものから消えます。 コンシューマーが何を読んだかはオフセットで管理されます。

まとめ

• Kafkaは「読んでも消えない分散ログ」。グループごとに同じデータを独立して読める
• 4.0以降はKRaftのみ。kafka-storage.sh format → kafka-server-start.sh で起動し、すべて --bootstrap-server で操作
• パーティションは並列度と順序の単位。同じキーは同じパーティションに入り、順序が保たれる
• Pythonは confluent-kafka。produce() 後の flush()、処理後の commit() が基本
• 迷ったら acks=all・レプリカ3・min.insync.replicas=2、配信保証はat-least-onceと重複に強い処理の組み合わせ
• 複数システムへの配信や読み直しならKafka、タスクの振り分けならRabbitMQやクラウドのキュー

株式会社RJCでは、Kafkaをはじめとするメッセージ基盤の仕組みを仲間と学び合っています。こうした知識を共有しながら、日々技術を磨いています。

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