KafkaとKarafkaで実現するRailsアプリのイベントストリーミング徹底解説

現代の企業は、大量のデータをリアルタイムに処理・共有し、そこからインサイトを得たり、より魅力的な顧客体験を創出したりすることが求められています。そのため、従来型のデータ処理の手法ではもはや時代に対応できなくなっています。

これを実現するには、膨大なデータをできるだけ高速に処理し、さらなる加工のために他のサービスへ送信する必要があります。そして、こうした素早い処理の中で重要になるのが「イベント発生時にコンシューマーへ通知する仕組み」です。これこそがイベントストリーミングが活躍する場面です。

本記事では、実際にGitHub上で公開しているサンプルリポジトリを使いながら解説を進めていきます。

イベントとは何か

イベントストリーミングの話に入る前に、まず「イベント」について整理しておきましょう。アプリケーション内で発生するイベントとは、ユーザーの操作プロセスに関連することもあれば、ビジネスに影響を与える単純なアクションであることもあります。

重要なのは、イベントが「状態の変化」を表すものであり、アプリケーションをどう変更するかという問題ではないという点です。具体例を挙げてみます。

• ユーザーがサービスにログインした
• 決済トランザクションが完了した
• ブロガーが記事を公開した

多くの場合、ひとつのイベントはさらに別のイベントを引き起こします。たとえば、ユーザーがサービスに登録すると、アプリは端末へ通知を送り、データベースにレコードを挿入し、ウェルカムメールを送信します。このように一連の連鎖が生まれるのです。

イベントストリーミングの基本

イベントストリーミングとは、データベースなどのイベントソースからデータをリアルタイムにキャプチャするためのパターンです。主な構成要素は以下のとおりです。

• Broker（ブローカー）: イベントを保存する役割を担うシステム
• Topic（トピック）: イベントのカテゴリー
• Producer（プロデューサー）: 特定のトピックに対してイベントを送信する側
• Consumer（コンシューマー）: イベントを読み取る側
• Events（イベント）: プロデューサーがコンシューマーへ伝えたいデータ本体

ここで必ず触れておきたいのがパブリッシュ/サブスクライブアーキテクチャ（pub/subパターン）です。イベントストリーミングはこのパターンの実装形態ですが、以下のような違いがあります。

• メッセージではなくイベントが流れる
• イベントは通常、時系列順に並べられる
• コンシューマーはトピック内の特定の時点からイベントを読み取れる
• イベントには時間的な永続性がある

処理の流れはこうなります。まずプロデューサーが新しいイベントをトピックにパブリッシュします（前述のとおり、トピックは特定種類のイベントを分類するためのものです）。次に、そのカテゴリーのイベントに関心を持つコンシューマーが該当トピックをサブスクライブします。最後にブローカーがトピックのコンシューマーを識別し、必要なイベントを提供します。

イベントストリーミングのメリット

• 疎結合
  パブリッシャーとコンシューマーは互いを知る必要がないため、依存関係がありません。また、イベント自体はアクションを指定しないため、同じイベントを受け取った複数のコンシューマーがそれぞれ異なる処理を行うことも可能です。

• 低レイテンシー
  イベントは疎結合であり、コンシューマーはいつでも利用できます。処理はミリ秒単位で行われることもあります。

• 独立性
  パブリッシャーとコンシューマーは独立しているため、異なるチームが同じイベントを別の目的や用途に使って開発を進めることができます。

• 耐障害性
  多くのイベントストリーミングプラットフォームはコンシューマーの障害への対処を支援します。たとえば、エラー発生時に読み取り位置を保存し、そこから再開することができます。

• リアルタイム処理
  フィードバックがリアルタイムに届くため、ユーザーは数分や数時間も待つことなく結果を確認できます。

• 高いパフォーマンス
  低レイテンシーのおかげで、イベントプラットフォームは大量のメッセージを捌けます。1秒あたり数千件のイベント処理も珍しくありません。

イベントストリーミングのデメリット

• 監視（モニタリング）
  一部のツールには十分なモニタリング機能が備わっておらず、DatadogやNew Relicなど、追加のツール導入が必要になる場合があります。

• 設定の複雑さ
  ツールによっては、経験豊富なエンジニアでも設定に苦労することがあります。パラメータが非常に多く、深い知識がないと適切に実装できないケースもあります。

• クライアントライブラリ
  Java以外の言語でKafkaを実装するのは容易ではありません。クライアントライブラリが最新でなかったり、不安定だったり、選択肢が限られていたりすることもあります。

イベントストリーミングで最も人気のあるツールのひとつがApache Kafkaです。このツールを使うと、必要なときに必要な場所でデータを送信・保存・取得できます。詳しく見ていきましょう。

Apache Kafkaとは

「Apache Kafkaはオープンソースの分散型イベントストリーミングプラットフォームであり、数千社もの企業によって、高性能なデータパイプライン、ストリーミング分析、データ統合、ミッションクリティカルなアプリケーションの構築に利用されています。」

Apache Kafkaはリアルタイムのログ伝送を目的として設計されているため、次のような要件を持つアプリケーションに最適です。

>コンポーネント間の信頼性の高いデータ交換
• アプリケーションの要件変化に応じたメッセージング負荷の分割
• データ処理のためのリアルタイム伝送

それでは、実際にRailsアプリケーションでKafkaを使ってみましょう。

RailsでKafkaを使う方法

RubyでKafkaを扱う有名なgemとしてはZendesk製のruby-kafkaがあります。非常に優秀ですが、すべての実装を手作業で行う必要があります。そこで登場するのが、ruby-kafkaをベースに構築された「フレームワーク」群です。設定や実行の手順も大幅に簡略化してくれます。

Karafkaは、Apache KafkaベースのRubyアプリケーション開発をシンプルにするためのフレームワークです。

Kafkaを動かすにはJavaのインストールが必要です。KafkaはScalaとJavaで書かれたアプリケーションでもあるため、Zookeeperのインストールも必須となります。

インストール手順の前に、Zookeeperについて簡単に説明しておきます。ZookeeperはKafkaにとって不可欠な中央集権型サービスで、新しいトピックの作成、ブローカーのクラッシュ、ブローカーの削除、トピックの削除などの変更が起きた際に通知を送ります。

主な役割はKafkaブローカーの管理、それぞれのメタデータ一覧の維持、ヘルスチェック機構の提供です。加えて、トピックの各パーティションにおけるリーダーブローカーの選出も支援します。

環境準備

macOSの場合：

次のコマンドでJavaとZookeeperをインストールします。

brew install java
brew install zookeeper

続いて、以下を実行してKafkaをインストールします。

brew install kafka

KafkaとZookeeperのインストールが完了したら、次のようにサービスを起動します。

brew services start zookeeper
brew services start kafka

Windows / Linuxの場合：

おおまかな手順は以下のとおりです。

1. Javaをインストールする
2. Zookeeperをダウンロードする

Railsアプリのセットアップ

いつものようにシンプルなRailsアプリを作成します。

rails new karafka_example

Gemfileにkarafka gemを追加しましょう。

gem 'karafka'

bundle install を実行してgemをインストールし、忘れずに以下のコマンドも実行してKarafka関連のファイルを生成します。

bundle exec karafka install

このコマンドにより、いくつかの重要なファイルが生成されます。ルートディレクトリの karafka.rb、app/consumers/application_consumer.rb、そして app/responders/application_responder.rb です。

Karafkaの初期化ファイル

karafka.rb は、Railsのconfigから分離されたイニシャライザのようなファイルです。ここでKarafkaアプリケーションの設定を行い、Railsのroutesに似たAPIでルーティングを定義できます。ただし、対象はコントローラーではなく「トピック」と「コンシューマー」です。

プロデューサーの実装

プロデューサーはイベントの生成を担当し、app/responders フォルダに配置します。ユーザー向けのシンプルなプロデューサーを作ってみましょう。

# app/responders/users_responder.rb

class UsersResponder < ApplicationResponder
  topic :users

  def respond(event_payload)
    respond_to :users, event_payload
  end
end

コンシューマーの実装

コンシューマーは、プロデューサーから送られたすべてのイベント/メッセージを読み取る役割です。ここでは受信したメッセージをログに出力するだけのシンプルなコンシューマーを作成します。

# app/consumers/users_consumer.rb

class UsersConsumer < ApplicationConsumer
  def consume
    Karafka.logger.info "New [User] event: #{params}"
  end
end

イベントの取得には params を使用します。ただし、イベントをバッチで読み取り、config.batch_fetching がtrueに設定されている場合は、params_batch を使用してください。

動作確認

Karafkaサービス（イベントを待ち受けるプロセス）を起動するには、コンソールで新しいタブを開き、Railsプロジェクトのディレクトリに移動して以下を実行します。

bundle exec karafka server

イベント成功時の挙動

次に、別のコンソールタブを開いてRailsプロジェクトに移動し、以下を入力します。

rails c

ここでレスポンダーを使ってイベントを作成してみましょう。

> UsersResponder.call({ event_name: "user_created", payload: { user_id: 1 } })

Railsコンソールを確認すると、イベント作成後に次のメッセージが表示されます。

Successfully appended 1 messages to users/0 on 192.168.1.77:9092 (node_id=0)
=> {"users"=>[["{\"event_name\":\"user_created\",\"payload\":{\"user_id\":1}}", {:topic=>"users"}]]}

Karafkaサービス側のタブでは、次のような出力が確認できます。

New [User] event: #<Karafka::Params::Params:0x00007fa76f0316c8>
Inline processing of topic users with 1 messages took 0 ms
1 message on users topic delegated to UsersConsumer
[[karafka_example] {}:] Marking users/0:1 as processed
[[karafka_example] {}:] Committing offsets: users/0:2
[[karafka_example] {}:] [offset_commit] Sending offset_commit API request 28 to 192.168.1.77:9092

もしメッセージのペイロードだけが必要なら、コンシューマーで params.payload を使うことで、次のように表示されます。

Params deserialization for users topic successful in 0 ms
New [User] event: {"event_name"=>"user_created", "payload"=>{"user_id"=>1}}
Inline processing of topic users with 1 messages took 1 ms
1 message on users topic delegated to UsersConsumer

イベント失敗時の挙動

次のコマンドで、email、first_name、last_name といった属性を持つUserモデルを作成してみましょう。

rails g model User email first_name last_name

続いてマイグレーションを実行します。

rails db:migrate

そして、次のようにバリデーションを追加します。

class User < ApplicationRecord
  validates :email, uniqueness: true
end

最後に、コンシューマーを次のように変更します。

class UsersConsumer < ApplicationConsumer
  def consume
    Karafka.logger.info "New [User] event: #{params.payload}"
    User.create!(params.payload['user'])
  end
end

それでは、同じメールアドレスを持つ2つのイベントを作成してみます。

UsersResponder.call({ event_name: "user_created", user: { user_id: 1, email: 'batman@mail.com', first_name: 'Bruce', last_name: 'Wayne' } } )

UsersResponder.call({ event_name: "user_created", user: { user_id: 2, email: 'batman@mail.com', first_name: 'Bruce', last_name: 'Wayne' } } )

最初のイベントは正常にデータベースへ登録されます。

New [User] event: {"event_name"=>"user_created", "user"=>{"user_id"=>1, "email"=>"batman@mail.com", "first_name"=>"Bruce", "last_name"=>"Wayne"}}
[[karafka_example] {users: 0}:] [fetch] Received response 2 from 192.168.1.77:9092
[[karafka_example] {users: 0}:] Fetching batches
[[karafka_example] {users: 0}:] [fetch] Sending fetch API request 3 to 192.168.1.77:9092
[[karafka_example] {users: 0}:] [fetch] Waiting for response 3 from 192.168.1.77:9092
  TRANSACTION (0.1ms)  BEGIN
  ↳ app/consumers/users_consumer.rb:14:in `consume'
  User Create (9.6ms)  INSERT INTO "users" ("user_id", "email", "first_name", "last_name", "created_at", "updated_at") VALUES ($1, $2, $3, $4, $5, $6) RETURNING "id"  [["user_id", "1"], ["email", "batman@mail.com"], ["first_name", "Bruce"], ["last_name", "Wayne"], ["created_at", "2021-03-10 04:29:14.827778"], ["updated_at", "2021-03-10 04:29:14.827778"]]
  ↳ app/consumers/users_consumer.rb:14:in `consume'
  TRANSACTION (5.0ms)  COMMIT
  ↳ app/consumers/users_consumer.rb:14:in `consume'
Inline processing of topic users with 1 messages took 70 ms
1 message on users topic delegated to UsersConsumer

しかし2つ目のイベントは失敗します。メールアドレスの一意性を保証するバリデーションがあるためです。既存のメールアドレスで別のレコードを登録しようとすると、次のようなログが出力されます。

New [User] event: {"event_name"=>"user_created", "user"=>{"user_id"=>2, "email"=>"batman@mail.com", "first_name"=>"Bruce", "last_name"=>"Wayne"}}
[[karafka_example] {users: 0}:] [fetch] Received response 2 from 192.168.1.77:9092
[[karafka_example] {users: 0}:] Fetching batches
[[karafka_example] {users: 0}:] [fetch] Sending fetch API request 3 to 192.168.1.77:9092
[[karafka_example] {users: 0}:] [fetch] Waiting for response 3 from 192.168.1.77:9092
  TRANSACTION (0.2ms)  BEGIN
  ↳ app/consumers/users_consumer.rb:14:in `consume'
  User Exists? (0.3ms)  SELECT 1 AS one FROM "users" WHERE "users"."email" = $1 LIMIT $2  [["email", "batman@mail.com"], ["LIMIT", 1]]
  ↳ app/consumers/users_consumer.rb:14:in `consume'
  TRANSACTION (0.2ms)  ROLLBACK
  ↳ app/consumers/users_consumer.rb:14:in `consume'
[[karafka_example] {users: 0}:] Exception raised when processing users/0 at offset 42 -- ActiveRecord::RecordInvalid: Validation failed: Email has already been taken

最終行に ActiveRecord::RecordInvalid: Validation failed: Email has already been taken というエラーが表示されていますね。ここで興味深いのは、Kafkaがこのイベントを何度も繰り返し処理しようとする点です。Karafkaサーバーを再起動しても、最後のイベントを再度処理しようとします。Kafkaはどこから再開すればいいのか、どうやって知っているのでしょうか？

コンソールを見ると、エラーの後に次のログが出ています。

[[karafka_example] {users: 0}:] Exception raised when processing users/0 at offset 42

これはどのオフセットまで処理したかを示しています。この場合はオフセット42です。つまり、Karafkaサービスを再起動すると、そのオフセットから処理を再開します。

[[karafka_example] {}:] Committing offsets with recommit: users/0:42
[[karafka_example] {users: 0}:] Fetching batches

Userモデルにメールアドレスのバリデーションがあるため、まだ失敗し続けます。ここで一度Karafkaサーバーを停止し、該当のバリデーションを削除またはコメントアウトしてからサーバーを再起動すると、イベントが正常に処理される様子を確認できます。

[[karafka_example] {}:] Committing offsets with recommit: users/0:42
[[karafka_example] {}:] [offset_commit] Sending offset_commit API request 5 to 192.168.1.77:9092
[[karafka_example] {}:] [offset_commit] Waiting for response 5 from 192.168.1.77:9092
[[karafka_example] {}:] [offset_commit] Received response 5 from 192.168.1.77:9092
Params deserialization for users topic successful in 0 ms
New [User] event: {"event_name"=>"user_created", "user"=>{"user_id"=>2, "email"=>"batman@mail.com", "first_name"=>"Bruce", "last_name"=>"Wayne"}}
  TRANSACTION (0.2ms)  BEGIN
  ↳ app/consumers/users_consumer.rb:14:in `consume'
  User Create (3.8ms)  INSERT INTO "users" ("user_id", "email", "first_name", "last_name", "created_at", "updated_at") VALUES ($1, $2, $3, $4, $5, $6) RETURNING "id"  [["user_id", "2"], ["email", "batman@mail.com"], ["first_name", "Bruce"], ["last_name", "Wayne"], ["created_at", "2021-03-10 04:49:37.832452"], ["updated_at", "2021-03-10 04:49:37.832452"]]
  ↳ app/consumers/users_consumer.rb:14:in `consume'
  TRANSACTION (5.5ms)  COMMIT
  ↳ app/consumers/users_consumer.rb:14:in `consume'
Inline processing of topic users with 1 messages took 69 ms
1 message on users topic delegated to UsersConsumer
[[karafka_example] {}:] Marking users/0:43 as processed

最後の行に Marking users/0:43 as processed というメッセージが表示されました。これでオフセット43まで処理が完了したことがわかります。

コールバック

Karafkaの便利な機能のひとつに、コンシューマーでコールバックを使える点があります。使い方は簡単で、モジュールをインクルードして定義するだけです。UsersConsumer を開いて、次のコードを追加してみましょう。

class UsersConsumer < ApplicationConsumer
  include Karafka::Consumers::Callbacks

  before_poll do
    Karafka.logger.info "*** Checking something new for #{topic.name}"
  end

  after_poll do
    Karafka.logger.info '*** We just checked for new messages!'
  end

  def consume
    Karafka.logger.info "New [User] event: #{params.payload}"
    User.create!(params.payload['user'])
  end
end

「Poll（ポーリング）」とは、現在のパーティションオフセットに基づいてレコードを取得する仕組みです。名前のとおり、before_poll と after_poll の各コールバックは、そのタイミングで実行されます。ここでは単にメッセージをログ出力しているだけですが、Karafkaサーバーのログで、フェッチ前とフェッチ後にそれぞれ出力される様子を確認できます。

*** Checking something new for users
[[karafka_example] {}:] No batches to process
[[karafka_example] {users: 0}:] [fetch] Received response 325 from 192.168.1.77:9092
[[karafka_example] {users: 0}:] Fetching batches
[[karafka_example] {users: 0}:] [fetch] Sending fetch API request 326 to 192.168.1.77:9092
[[karafka_example] {users: 0}:] [fetch] Waiting for response 326 from 192.168.1.77:9092
*** We just checked for new messages!

ハートビート

ハートビートとは、コンシューマーとして「私たちが生きていること」をKafkaに伝える手段です。ハートビートがなければ、Kafkaはそのコンシューマーがダウンしたと判断します。

Karafkaには、一定間隔でハートビートを送るためのデフォルト設定があります。それが kafka.heartbeat_interval で、デフォルト値は10秒です。このハートビートはKarafkaサーバーのログで確認できます。

*** Checking something new for users
[[karafka_example_example] {}:] Sending heartbeat...
[[karafka_example_example] {}:] [heartbeat] Sending heartbeat API request 72 to 192.168.1.77:9092
[[karafka_example_example] {}:] [heartbeat] Waiting for response 72 from 192.168.1.77:9092
[[karafka_example_example] {}:] [heartbeat] Received response 72 from 192.168.1.77:9092
*** We just checked for new messages!

Sending heartbeat... というログによって、Kafkaは私たちが生存しており、コンシューマーグループの有効なメンバーであることを認識します。これにより、引き続きレコードを消費できる状態が保たれます。

コミット

あるオフセットまで消費済みとしてマークすることを「オフセットのコミット」と呼びます。Kafkaでは、offsetsトピックと呼ばれる内部トピックに書き込むことでオフセットのコミットを記録します。メッセージは、そのオフセットがoffsetsトピックにコミットされて初めて「消費済み」とみなされます。

Karafkaには、このコミットを自動的に定期実行するための設定があります。それが kafka.offset_commit_interval で、デフォルト値は10秒です。この設定により、Karafkaは10秒ごとにオフセットコミットを行います。Karafkaサーバーのログでその様子を確認できます。

*** Checking something new for users
[[karafka_example] {}:] No batches to process
[[karafka_example] {users: 0}:] [fetch] Received response 307 from 192.168.1.77:9092
[[karafka_example] {users: 0}:] Fetching batches
[[karafka_example] {users: 0}:] [fetch] Sending fetch API request 308 to 192.168.1.77:9092
[[karafka_example] {users: 0}:] [fetch] Waiting for response 308 from 192.168.1.77:9092
[[karafka_example] {}:] Committing offsets: users/0:44
[[karafka_example] {}:] [offset_commit] Sending offset_commit API request 69 to 192.168.1.77:9092
[[karafka_example] {}:] [offset_commit] Waiting for response 69 from 192.168.1.77:9092
[[karafka_example] {}:] [offset_commit] Received response 69 from 192.168.1.77:9092
*** We just checked for new messages!

Committing offsets: users/0:44 は、どのオフセットをコミットしたかを示しています。この例では、トピック0のオフセット44までコミット可能だとKafkaに伝えています。この仕組みのおかげで、万が一サービスに障害が発生しても、Karafkaはそのオフセットからイベント処理を再開できるのです。

まとめ

イベントストリーミングを活用することで、より迅速な処理、データの有効活用、そして優れたユーザー体験の設計が可能になります。実際、多くの企業がこのパターンを採用してサービス間の連携を行い、さまざまなイベントにリアルタイムで対応できるようにしています。前述のとおり、Railsで使える選択肢はKarafka以外にもあります。本記事で基礎は身についたはずなので、ぜひいろいろと試してみてください。

参考資料

• https://kafka.apache.org/
• https://github.com/karafka/karafka
• https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern