エラーメッセージRedis Streamsを使用して分散メッセージングシステムを構築していると、この問題に直面することがよくあります。このエラーは、システム内に既に存在するコンシューマーグループをセットアップしようとした瞬間に発生します。
(error) BUSYGROUP Consumer Group name already exists
このエラーはデータベースが破損しているわけではありません。単に、送信した XGROUP CREATE コマンドが、既存のリソースを重複して作成しようとしていることをRedisが伝えているだけです。
エラーが発生する理由XGROUP CREATE コマンドには「べき等性」がありません。Redisは、グループ作成の各呼び出しがユニークな名前であることを期待しています。同じグループを「二重に作成」しようとすると、コマンドは失敗します。これは通常、以下の3つのシナリオで発生します。
- サービスの再起動: アプリケーションの起動時に、毎回グループを作成しようとする初期化スクリプトが実行されている。- 水平スケーリング: ワーカーのインスタンスを10個同時に起動すると、それらすべてが同じコンシューマーグループを作成しようと競合します。最初の1つは成功しますが、残りは
BUSYGROUPエラーでクラッシュします。- 古い環境設定: 開発者がデバッグ中にredis-cliを使って手動でグループを作成してしまい、自動デプロイスクリプトがそれに衝突している。## ステップバイステップの解決策### ステップ1: 既存のグループを確認するコードを変更する前に、Redisインスタンスで実際に何が実行されているかを確認してください。XINFO GROUPSコマンドを使用して、ストリームに関連付けられているすべてのグループを表示します。mystreamは実際のキー名に置き換えてください。
XINFO GROUPS mystream
このコマンドを実行すると、グループのリスト、現在のコンシューマー数、保留中のメッセージ数が返されます。リストに目的のグループ名があれば、競合が確認されたことになります。ERR no such key エラーが表示される場合は、ストリーム自体がまだ存在していません。
ステップ2: べき等なロジックを実装する本番環境では、作成前に存在チェックを行わないでください。そのパターンでは、2つのワーカーが同時に「グループが存在しない」と判断し、両方が作成を試みるという「レースコンディション（競合状態）」が発生します。代わりに、グループの作成を試み、既に存在する場合はエラーをキャッチするようにします。
Pythonでの実装例 (redis-py):
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
stream_name = "events_log"
group_name = "analytics_workers"
try:
# '$' は新しいメッセージのみを読み取ります。履歴全体を処理する場合は '0' を使用してください。
r.xgroup_create(stream_name, group_name, id='$', mkstream=True)
print("成功: コンシューマーグループが作成されました。")
except redis.exceptions.ResponseError as e:
if "BUSYGROUP" in str(e):
print("グループは既に存在します。作業を再開します。")
else:
raise e
Node.jsでの実装例 (ioredis):
const Redis = require('ioredis');
const redis = new Redis();
async function ensureGroup() {
try {
// MKSTREAM はストリームが存在しない場合に自動的に作成します
await redis.xgroup('CREATE', 'orders', 'processor-group', '$', 'MKSTREAM');
} catch (err) {
if (err.message.includes('BUSYGROUP')) {
return; // エラーを無視: グループは既に存在しています
}
throw err;
}
}
ステップ3: グループオフセットの更新グループを削除したくないが、読み取り開始位置を変更したい場合があります（例：過去5,000件のメッセージを再処理したい場合）。このためにグループを破棄する必要はありません。代わりに XGROUP SETID を使用します。
# ストリームの最初から開始するようにグループを変更します
XGROUP SETID mystream mygroup 0
完全にやり直す必要がある場合は XGROUP DESTROY を使用しますが、そのグループのすべての保留中メッセージデータが消去されるため、注意してください。
XGROUP DESTROY mystream mygroup