メニュー
Message queue patterns with RabbitMQ, Redis Streams, BullMQ, dead letter queues, exactly-once delivery, and event-driven architectures
デフォルトでは、最初にソースを確認する Prompt が選択されています。直接コマンドに切り替えるか、ローカルコピーをダウンロードすることもできます。
インストールを決める前に、SKILL.md と SkillsMP に表示されている付属ファイルをお読みください。
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
直接コマンドでは確認用 Prompt が省略されます。実行前にソースを確認してください。
npx skills add https://github.com/InugamiDev/ultrathink-oss --skill message-queuesコマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか？SkillsMP が現在取得できるファイルをダウンロードできます。
Unified design foundations — design system architecture, tokens, component specs, visual principles, creative vision, figma integration, plus brand design system loader (66 real brands via DESIGN.md). Absorbs design, design-system, design-systems, design-principles, design-router, creative-vision, figma, design-md.
Render, summarize, and present markdown documents and structured content in multiple output modes
Ultra UI skill - combines Google's DESIGN.md spec (machine-readable design tokens) with the ui-ux-pro-max knowledge base (91 styles, 161 palettes, 73 font pairings, 161 products, 104 UX guidelines, 25 chart types). Generates lint-clean DESIGN.md files, validates token references and WCAG contrast, exports Tailwind/DTCG tokens, and diffs design systems version-over-version.
SOC 職業分類に基づく
SKILL.md を表示中
| name | message-queues |
| description | Message queue patterns with RabbitMQ, Redis Streams, BullMQ, dead letter queues, exactly-once delivery, and event-driven architectures |
| layer | domain |
| category | backend |
| triggers | ["message queue","rabbitmq","redis streams","bullmq","bull","event driven","pub sub","dead letter queue","job queue","task queue","background jobs"] |
| inputs | ["Async processing requirements","Event-driven architecture needs","Job scheduling specifications"] |
| outputs | ["Queue topology designs","Producer/consumer implementations","Retry and dead letter configurations"] |
| linksTo | ["redis","microservices","nodejs","python","golang"] |
| linkedFrom | ["error-handling","logging","monitoring"] |
| preferredNextSkills | ["redis","microservices","nodejs"] |
| fallbackSkills | ["websockets","redis"] |
| riskLevel | low |
| memoryReadPolicy | selective |
| memoryWritePolicy | none |
| sideEffects | [] |
Provide expert-level guidance on message queue patterns, including RabbitMQ exchange topologies, Redis Streams consumer groups, BullMQ job processing, dead letter queues, retry strategies, exactly-once semantics, and event-driven architecture design.
| Technology | Best For | Ordering | Persistence | Throughput |
|---|---|---|---|---|
| BullMQ | Job queues in Node.js apps | FIFO per queue | Redis-backed | High |
| Redis Streams | Event logs, lightweight pub/sub | Per-stream | Configurable | Very high |
| RabbitMQ | Complex routing, multi-consumer | Per-queue | Durable | High |
| Kafka | Event sourcing, high-volume streaming | Per-partition | Durable | Extreme |
| SQS | Serverless, AWS-native | Best-effort | Managed | High |
import { Queue, Worker, QueueScheduler, FlowProducer } from 'bullmq';
import IORedis from 'ioredis';
const connection = new IORedis({
host: process.env.REDIS_HOST,
port: 6379,
maxRetriesPerRequest: null, // Required by BullMQ
});
// Define typed job data
{
: ;
: ;
: ;
: <, >;
}
{
: ;
: ;
}
emailQueue = <, >(, {
connection,
: {
: ,
: {
: ,
: ,
},
: { : * },
: { : * * },
},
});
() {
emailQueue.(, data, {
: opts?.,
: opts?.,
: ,
});
}
emailWorker = <, >(
,
(job) => {
job.();
job.();
html = (job.., job..);
job.();
result = mailer.({
: job..,
: job..,
html,
});
job.();
{ : result., : ().() };
},
{
connection,
: ,
: {
: ,
: ,
},
}
);
emailWorker.(, {
logger.({ : job., result }, );
});
emailWorker.(, {
logger.({ : job?., : err., : job?. }, );
});
emailQueue.(, { : }, {
: {
: ,
: ,
},
});
flow = ({ connection });
flow.({
: ,
: ,
: { : },
: [
{ : , : , : { : } },
{ : , : , : { : } },
],
});
import Redis from 'ioredis';
const redis = new Redis();
// Producer: Add events to stream
async function publishEvent(stream: string, event: Record<string, string>) {
// MAXLEN ~ 10000 trims stream approximately (more efficient than exact)
return redis.xadd(stream, 'MAXLEN', '~', '10000', '*', ...Object.entries(event).flat());
}
await publishEvent('orders', {
type: 'order.created',
orderId: '12345',
userId: 'user-1',
total: '9999',
timestamp: Date.now().toString(),
});
// Consumer group: Reliable, multi-consumer processing
const GROUP = 'order-processors';
const CONSUMER = `worker-${process.pid}`;
const STREAM = 'orders';
// Create consumer group (idempotent with MKSTREAM)
try {
await redis.xgroup('CREATE', STREAM, GROUP, '0', 'MKSTREAM');
} catch {
// Group already exists
}
// Process loop
async function consumeStream() {
while (true) {
// Read new messages (> means undelivered to this consumer)
const entries = await redis.xreadgroup(
'GROUP', GROUP, CONSUMER,
'COUNT', '10',
'BLOCK', '5000', // Block for 5s if no messages
'STREAMS', STREAM, '>'
);
if (!entries) continue;
for (const [, messages] of entries) {
for (const [id, fields] of messages) {
try {
const event = Object.fromEntries(
fields.reduce((acc, val, i) => {
if (i % 2 === 0) acc.push([val, fields[i + 1]]);
return acc;
}, [] as [string, string][])
);
await processEvent(event);
// Acknowledge successful processing
await redis.xack(STREAM, GROUP, id);
} catch (error) {
logger.error({ id, error }, 'Failed to process event');
// Message will be re-delivered (pending entries list)
}
}
}
}
}
// Claim stale messages (for crash recovery)
async function claimStaleMessages() {
// Find messages pending for more than 5 minutes
const pending = await redis.xpending(STREAM, GROUP, '-', '+', '100');
for (const [id, consumer, idleTime] of pending) {
if (idleTime > 5 * 60 * 1000) {
const claimed = await redis.xclaim(
STREAM, GROUP, CONSUMER,
5 * 60 * 1000, // Min idle time
id
);
logger.info({ id, previousConsumer: consumer }, 'Claimed stale message');
}
}
}
interface RetryPolicy {
maxAttempts: number;
backoff: 'fixed' | 'exponential';
baseDelay: number;
maxDelay: number;
}
const DEFAULT_RETRY: RetryPolicy = {
maxAttempts: 5,
backoff: 'exponential',
baseDelay: 1000,
maxDelay: 60_000,
};
async function processWithDLQ<T>(
message: T,
processor: (msg: T) => Promise<void>,
dlqQueue: Queue,
policy: RetryPolicy = DEFAULT_RETRY,
attempt = 1,
) {
try {
await processor(message);
} catch (error) {
if (attempt >= policy.maxAttempts) {
// Send to Dead Letter Queue
await dlqQueue.add('dead-letter', {
originalMessage: message,
error: error instanceof Error ? error.message : String(error),
attempts: attempt,
failedAt: new Date().toISOString(),
});
logger.error({ message, attempt, error }, 'Message moved to DLQ');
return;
}
// Calculate retry delay
const delay = policy.backoff === 'exponential'
? Math.min(policy.baseDelay * Math.pow(2, attempt - 1), policy.maxDelay)
: policy.baseDelay;
logger.warn({ message, attempt, delay, error }, 'Retrying message');
// Re-queue with delay
setTimeout(() => processWithDLQ(message, processor, dlqQueue, policy, attempt + 1), delay);
}
}
// Prevent duplicate processing using an idempotency key
class IdempotentConsumer {
constructor(private redis: Redis, private ttl: number = 86400) {}
async process<T>(
messageId: string,
handler: () => Promise<T>,
): Promise<T | null> {
// Try to acquire processing lock
const acquired = await this.redis.set(
`idempotency:${messageId}`,
'processing',
'EX', this.ttl,
'NX' // Only set if not exists
);
if (!acquired) {
// Already processed or in progress
const status = await this.redis.get(`idempotency:${messageId}`);
if (status === 'completed') {
logger.debug({ messageId }, 'Duplicate message, already processed');
return null;
}
// Still processing by another consumer -- skip
return null;
}
try {
const result = await handler();
await this.redis.set(`idempotency:${messageId}`, 'completed', 'EX', this.ttl);
return result;
} catch (error) {
await this.redis.del(`idempotency:${messageId}`);
throw error;
}
}
}
| Pitfall | Impact | Fix |
|---|---|---|
| Assuming exactly-once delivery | Duplicate processing | Idempotent consumers with deduplication keys |
| No dead letter queue | Poison messages block the queue | Configure DLQ with max retry policy |
| Unbounded retry loops | Infinite resource consumption | Set max attempts with exponential backoff |
| Large message payloads | Memory pressure, slow delivery | Store payload in S3/DB, pass reference in message |
| Not acknowledging messages | Message re-delivery storm | Always ACK after successful processing |
| Synchronous processing of dependent events | Cascading failures | Use separate queues, saga pattern |
| No monitoring/alerting | Silent queue buildup | Monitor queue depth, consumer lag, failure rate |