フィットネスのイベント駆動アーキテクチャ：Node.jsとRabbitMQでワークロードを非同期処理
Node.jsとRabbitMQを使用して、スケーラブルで堅牢なフィットネスアプリバックエンドを構築する方法を説明します。イベント駆動アーキテクチャでサービスを分離し、高速な非同期ワークロード処理を実現します。
このページの内容
fitnessアプリ使ったことありますか？1時間走っただけで.Badgesがすぐに付与されて、週間リーダーボードも瞬時に更新される——そんな裏側を体験したことがあるなら、まさに非同期処理の可能性を身をもって体験したことになるでしょう。
「ワークアウトを保存」中のスピナーが消える頃には、これらすべてが完了しているなら、あなたはすでに非同期処理の魔法を見ている状態です。
従来のモノリシックアーキテクチャでは、ユーザーがワークアウトを投稿すると、サーバーは複数のタスクを順番に実行する必要があります：データの検証、データベースへの保存、実績の計算、ユーザーの統計更新、リーダーボードの更新。高負荷時にはAPIの応答が遅くなり、ユーザー体験が著しく低下します。あるステップが失敗すると、プロセス全体をロールバックしなければならないこともあります。
このチュートリアルでは、Node.jsとRabbitMQを使って、シンプルで強力なイベント駆動型システムを構築し、この問題を解決します。2つのマイクロサービスを作成します：
workout-service：ユーザーからワークアウトデータを受け取り、メッセージキューに即座に送信するExpress APIです。このサービスの唯一の役割は、データ受領において高速で信頼性があることです。processing-service：キューからの新しいワークアウトメッセージをリスンするバックグラウンドワーカーで、統計の計算や実績の確認などの重い処理を実行し、結果をログに記録します。
この疎結合なアプローチにより、アプリケーションはよりスケーラブルで堅牢になり、保守も容易になります。✨
前提条件：
- Node.jsとnpm（またはyarn）がインストール済みであること。
- DockerとDocker Compose（RabbitMQを簡単に実行するため）。
- JavaScript（async/await）とExpress.jsの基本的な理解。
問題の理解：モノリシックなボトルネック
フィットネッシュアプリが大成功を収めると想像してください。何千人ものユーザーが同時にワークアウトを完了しています。単一のAPIエンドポイントが今、こんな状況です：
- 高レイテンシ： ユーザーは読み込みスピナーを見つめながら何秒も待ち、すべてのワークアウト後の計算が終わるのを待っています。
- 信頼性の低さ： 標高gainを計算するために使っているサードパーティサービスがダウンすると、チェーンの一部が壊れているため、誰もワークアウトを保存できなくなります。
- スケーラビリティの問題： ワークアウトのデータ取り込みは軽量ですが、実績計算はCPU集約的です。処理負荷を捌くためだけにアプリケーション全体をスケールしなければならず、非効率でコストもかかります。
メッセージキューを使ったイベント駆動型アーキテクチャは、これらの問題を解決します。workout-serviceはシンプルにワークアウトを受け入れ、RabbitMQキューにイベント（メッセージ）をPublishします。これは非常に高速な操作です。processing-serviceは自分のペースでこれらのイベントを消費し、独立してスケールできます。処理サービスが失敗しても、メッセージはキューに残り、サービスが回復すると処理される準備が整います。
イベント駆動型ワークアウト処理アーキテクチャ
次の図は、非同期ワークアウト処理フローを示しています：
前提条件：環境のセットアップ
まず、プロジェクト構造と依存関係を整備しましょう。
workout-serviceとprocessing-serviceという2つのパッケージを持つモノレポを作成します。
mkdir fitness-app-backend
cd fitness-app-backend
mkdir packages
cd packages
mkdir workout-service
mkdir processing-service
1. DockerによるRabbitMQ
開発用にRabbitMQを実行する一番簡単な方法はDockerです。fitness-app-backendディレクトリのルートにdocker-compose.ymlファイルを作成します。
# docker-compose.yml
version: '3.8'
services:
rabbitmq:
image: "rabbitmq:3.13-management"
ports:
- "5672:5672" # AMQP port for our services
- "15672:15672" # Management UI
environment:
- RABBITMQ_DEFAULT_USER=user
- RABBITMQ_DEFAULT_PASS=password
次に、実行します：
docker-compose up -d
RabbitMQ Management UIにはhttp://localhost:15672でアクセスでき、user/passwordでログインできます。
2. サービスの依存関係
workout-serviceの場合：
cd packages/workout-service
npm init -y
npm install express amqplib
processing-serviceの場合：
cd ../processing-service
npm init -y
npm install amqplib
Node.jsでRabbitMQとやり取りするために、人気のamqplibライブラリを使用しています。
RabbitMQプロデューサーを使ったワークアウト取り込みAPIの構築
workout-serviceは単一のpurposeを持つExpress APIです：ワークアウトデータを受け取り、RabbitMQキューにPublishします。
実施内容
完了したワークアウトを表すJSONペイロードを受け取るPOSTリクエストのエンドポイント/workoutを作成します。このサービスはRabbitMQに接続し、workout_jobsという名前のキューにワークアウトデータを送ります。
実装
packages/workout-serviceの中にindex.jsというファイルを作成します。
// packages/workout-service/index.js
const express = require('express');
const amqp = require('amqplib');
const app = express();
app.use(express.json());
const RABBITMQ_URL = 'amqp://user:password@localhost:5672';
const QUEUE_NAME = 'workout_jobs';
let channel; // RabbitMQ channel
// --- RabbitMQ Connection ---
async function connectRabbitMQ() {
try {
const connection = await amqp.connect(RABBITMQ_URL);
channel = await connection.createChannel();
await channel.assertQueue(QUEUE_NAME, { durable: true }); // Durable queue
console.log('✅ Connected to RabbitMQ and workout_jobs queue is ready.');
} catch (error) {
console.error('❌ Failed to connect to RabbitMQ', error);
process.exit(1); // Exit if connection fails
}
}
// --- API Endpoint ---
app.post('/workout', (req, res) => {
const workoutData = req.body;
// Basic validation
if (!workoutData || !workoutData.type || !workoutData.duration) {
return res.status(400).send({ message: 'Invalid workout data' });
}
// Send to queue
const message = Buffer.from(JSON.stringify(workoutData));
channel.sendToQueue(QUEUE_NAME, message, { persistent: true }); // Persistent message
console.log(`📥 Workout data sent to queue:`, workoutData);
res.status(202).send({
message: 'Workout received and is being processed.',
});
});
// --- Start Server ---
const PORT = 3000;
app.listen(PORT, async () => {
await connectRabbitMQ();
console.log(`🚀 Workout service listening on port ${PORT}`);
});
動作の仕組み
- RabbitMQ接続：
connectRabbitMQ関数が接続を確立し、チャンネルを作成します。次にdurable: trueオプションでassertQueueを呼び出します。これにより、RabbitMQの再起動後もキューが存続することが保証されます。 - Expressエンドポイント：
/workoutエンドポイントはワークアウトデータを受け取りent. - メッセージのPublish： JSONワークアウトデータをBufferに変換し、
channel.sendToQueueを使用します。{ persistent: true }オプションはRabbitMQにメッセージをディスクに保存するよう指示し、サーバーがクラッシュしても失われません。 - HTTP 202 Accepted：
202 Acceptedステータスコードを返します。これは重要です——クライアントに「リクエストを受け取り、処理します」と伝え、処理が完了するのを待たせません。
ワークアウト処理用バックグラウンドワーカーの構築
このサービスはバックグラウンドワーカーです。SameRabbitMQキューに接続し、メッセージをリスンして「処理」します。
実施内容
ワーカーはworkout_jobsキューからワークアウトメッセージを1つずつpullします。各ワークアウトについて、いくつかの統計を計算し、シンプルな実績を確認する処理をシミュレートします。
実装
packages/processing-serviceの中にindex.jsというファイルを作成します。
// packages/processing-service/index.js
const amqp = require('amqplib');
const RABBITMQ_URL = 'amqp://user:password@localhost:5672';
const QUEUE_NAME = 'workout_jobs';
async function startWorker() {
try {
const connection = await amqp.connect(RABBITMQ_URL);
const channel = await connection.createChannel();
await channel.assertQueue(QUEUE_NAME, { durable: true });
// This ensures the worker only gets one message at a time
channel.prefetch(1);
console.log(`[ P ] Waiting for workouts in ${QUEUE_NAME}. To exit press CTRL+C`);
channel.consume(QUEUE_NAME, (msg) => {
if (msg !== null) {
const workoutData = JSON.parse(msg.content.toString());
console.log(`[ → ] Received workout:`, workoutData);
// Simulate processing
processWorkout(workoutData);
// Acknowledge the message
channel.ack(msg);
console.log('[ ✔ ] Done processing. Acknowledged message.');
}
}, { noAck: false }); // `noAck: false` requires us to manually acknowledge
} catch (error) {
console.error('❌ Failed to start worker', error);
}
}
function processWorkout(data) {
// 1. Calculate stats
const caloriesBurned = data.duration * 8.5; // Simple heuristic
console.log(`[ ⚙️ ] Processing: Calculated ${caloriesBurned.toFixed(2)} calories burned.`);
// 2. Check for achievements
if (data.type === 'Running' && data.distance >= 5) {
console.log(`[ 🏆 ] Achievement Unlocked: "5K Runner"!`);
// In a real app, you would save this to the database.
}
if (data.duration > 60) {
console.log(`[ 🏆 ] Achievement Unlocked: "Hour of Power"!`);
}
// Simulate a delay for processing
// In a real app, this could be database calls, API requests, etc.
const processingTime = Math.random() * 2000 + 500; // 0.5 to 2.5 seconds
const start = Date.now();
while (Date.now() - start < processingTime) {
// blocking for demo purposes
}
}
startWorker();
動作の仕組み
- 接続とキュー： ワーカーは同じ永続キュー`workout_jobsに接続します。
channel.prefetch(1)：これはワーカーにとって重要な設定です。RabbitMQに、現在のメッセージを処理して確認応答するまで新しいメッセージを送らないよう指示します。これにより、ワーカーが圧倒されるのを防ぎます。channel.consume：これはメッセージのリスニングを開始します。コールバック関数は各メッセージに対して実行されます。noAck: false：確認応答をmanualに明示的に設定します。これは信頼性にとって重要です。ワーカーが処理途中でクラッシュした場合、RabbitMQはメッセージが確認応答されなかったことを見て、再キューして別の（または同じ）ワーカーが再試行します。processWorkout()：この関数にはビジネスロジックが含まれています。実際のアプリケーションで実行するより複雑な計算のプレースホルダーです。channel.ack(msg)：processWorkoutが完了したら、ack(msg)を呼び出してRabbitMQに「このメッセージは正常に処理されました。安全に削除して構いません」と伝えます。
すべてをまとめる
では、システムを実際に見てみましょう！
-
RabbitMQを起動：
bash docker-compose up -d -
処理サービス（コンシューマー）を起動： ターミナルを開き、
packages/processing-serviceに移動して実行：bash node index.js次のような表示が見えるはずです：
[ P ] Waiting for workouts in workout_jobs... -
ワークアウトサービスを起動（プロデューサー）： 2番目のターミナルを開き、
packages/workout-serviceに移動して実行：bash node index.js次のような表示が見えるはずです：
🚀 Workout service listening on port 3000 -
ワークアウトを送信： 3番目のターミナル（またはPostmanなどのツールを使用して）、APIにPOSTリクエストを送信します。
実績が解除されるワークアウトのサンプル
curlコマンドです：bash curl -X POST http://localhost:3000/workout \ -H "Content-Type: application/json" \ -d '{ "userId": "user-123", "type": "Running", "duration": 45, "distance": 5.2, "timestamp": "2025-12-10T10:00:00Z" }'即座にレスポンスが来るはずです：
{"message":"Workout received and is being processed."}次に、他のターミナルを見てください！
- ワークアウトサービスの出力：
📥 Workout data sent to queue: { userId: 'user-123', type: 'Running', ... } - 処理サービスの出力：
[ → ] Received workout: { userId: 'user-123', type: 'Running', ... } [ ⚙️ ] Processing: Calculated 382.50 calories burned. [ 🏆 ] Achievement Unlocked: "5K Runner"! [ ✔ ] Done processing. Acknowledged message.
- ワークアウトサービスの出力：
成功です！ワークアウトを非同期で正常に処理しました。
セキュリティとパフォーマンスの考慮事項
- 入力検証： 現在の検証は最小限です。実際のアプリでは、
JoiやZodなどのライブラリを使用して、ワークアウトデータがキューに届く前にworkout-serviceで徹底的な検証を行うべきです。 - エラー処理：
processWorkoutが失敗したら怎么办？try...catchブロックでラップする必要があります。エラーが発生した場合、nack（否定確認応答）を選択してメッセージを破棄するか、再キューするかを選択できます。永続的な失敗については、 문제のあるメッセージを後で検査するために保存する**デッドレターキュー（DLQ）**のセットアップを検討してください。 - スケーラビリティ：
workout_jobsキューが伸び始めたら、より多くの処理能力が必要というサインです。このアーキテクチャの利点は、workout-serviceに触れずにprocessing-serviceのインスタンスを単に追加できることです。RabbitMQは利用可能なすべてのコンシューマーにメッセージを分散します。
代替アプローチ
- AWS SQS（Simple Queue Service）： 本番環境では、SQSのような管理サービスがより良い選択であることが多いです。非常にスケーラブルで耐久性に優れており、基盤となるサーバーを管理する必要がありません。コンセプト（キュー、プロデューサー、コンシューマー）は非常に似ています。
- Kafka： 高吞吐量のイベントストリーミング（ワークアウト中のリアルタイムの位置情報追跡など）には、ApacheKafkaのようなストリーム処理プラットフォームの方が適しているかもしれません。リアルタイムで海量のデータを処理하도록設計されています。
結論
フィットネッシュワークアウトを処理するための疎結合でイベント駆動型のシステムを正常に構築しました。メッセージキューを導入することで、アプリケーションをより堅牢でスケーラブルにし、スナップなユーザー体験を提供できるようになりました。workout-serviceは軽量で高速を保ち、processing-serviceは独立してスケールして任意の負荷を処理できます。
このパターンは非常に強力で、数え切れないほどの本番システムで使用されていますレポートの生成、ビデオのトランスコーディング、サードパーティAPIの呼び出しなど、アプリケーションの他の長時間実行タスクに今すぐ適用できます。
パフォーマンスへの影響：イベント駆動型アーキテクチャは、バックグラウンド処理を伴う操作のAPI応答時間を70〜85%短縮します。メッセージキューを使用するアプリケーションは、処理サービスの障害中に99.9%の上位率をを実現し、水平スケーリングの効率性によりコストを50〜70%削減します。疎結合なサービスは独立したスケーリングを可能にし、その結果、マイクロサービスベースのチームではリソース利用率が40%向上し、新機能開発が60%高速化します。
次のステップ：
- 失敗したワークアウトメッセージ用のデッドレターキュー（DLQ）を実装する。
- より複雑な実績ロジックを追加する。
- 処理されたワークアウトの結果をデータベース（MongoDBやPostgreSQLなど）に保存する。
その他のバックエンドアーキテクチャパターンについては、**KubernetesでのNode.jsスケーリングまたはNode.jsとRedisでのリアルタイムリーダーボード構築をご覧ください。イベント駆動型システムの詳細については、AWS LambdaとKinesisでのIoTデータ処理**をご覧ください。
リソース
- amqplib公式ドキュメント： https://www.squaremobius.net/amqp.node/
- RabbitMQチュートリアル： https://www.rabbitmq.com/getstarted.html
- オープンソース運動データセット： https://github.com/wrkout/exercises.json
クイックリファレンス
| 側面 | 主要情報 |
|---|---|
| トピック | 開発 |
| 関連性 | 患者教育と認識 |
| アクション | 医療提供者と相談してください |
| 頻度 | ケアチームのおすすめに従ってください |
| 証拠 | 現在の臨床ガイドラインに基づいています |
よくある質問
Q: これが自分に該当するかどうかはどうやってわかりますか？ A: 医療提供者と個人的および家族のがん歷について話し、個々のリスク要因を評価し、適切な次のステップを決定してください。
Q: 医師に何を聞くべきですか？ A: 良い質問例：「私の結果の意味は何ですか？」「私オプションは何がありますか？」「どんなライフスタイルの変更が役立ちますか？」「いつフォローアップすべきですか？」
Q: 助けることができるライフスタイルの変更はありますか？ A: ほとんどの健康状態は、基本的な対策から恩恵を受けます：バランスの取れた栄養、定期的な身体活動（週150分以上）、質のいい睡眠（7〜9時間）、ストレス管理、タバコの回避。
免责声明
本記事に記載されているアルゴリズムと手法は、技術教育の目的のみを目的としています。臨床的検証を受けておらず、医学的診断や治療決定に使用すべきではありません。医学的アドバイスについては、常に資格のある医療専門家にご相談ください。