队列生产者实例的工厂函数

const emailQueue = new Queue('email', {
  connection: {
    host: 'localhost',
    port: 6379,
  }
});

const imageQueue = new Queue('image', {
  connection: {
    host: 'localhost',
    port: 6379,
  }
});

const reportQueue = new Queue('report', {
  connection: {
    host: 'localhost',
    port: 6379,
  }
});

// queue-factory.ts
import { Queue } from 'bullmq';

const defaultConnection = {
  host: process.env.REDIS_HOST || 'localhost',
  port: parseInt(process.env.REDIS_PORT || '6379'),
};

export function createQueue(name: string) {
  return new Queue(name, {
    connection: defaultConnection,
  });
}

const emailQueue = createQueue('email');
const imageQueue = createQueue('image');
const reportQueue = createQueue('report');

// ❌ 这样做看似节省资源,但有隐患
let emailQueue: Queue | null = null;

export function getEmailQueue() {
  if (!emailQueue) {
    emailQueue = new Queue('email', { connection: defaultConnection });
  }
  return emailQueue;
}

import { Queue, QueueOptions } from 'bullmq';

export type QueueName = 'email' | 'image' | 'report';

export function createQueue<T = any>(
  name: QueueName,
  options?: Partial<QueueOptions>
): Queue<T> {
  return new Queue<T>(name, {
    connection: defaultConnection,
    ...options,
  });
}

const validQueue = createQueue('email');     // ✅ 通过
const invalidQueue = createQueue('unknown'); // ❌ 类型错误

interface EmailJob {
  to: string;
  subject: string;
  body: string;
}

const emailQueue = createQueue<EmailJob>('email');

await emailQueue.add('send', {
  to: 'user@example.com',
  subject: 'Welcome',
  body: 'Hello!',
});

// ❌ 类型错误:缺少必需字段
await emailQueue.add('send', {
  to: 'user@example.com',
});

// test-helpers.ts
import { Queue } from 'bullmq';

export function createTestQueue(name: string) {
  return new Queue(name, {
    connection: {
      host: 'localhost',
      port: 6380,  // 测试用的独立 Redis 实例
    },
  });
}

import { Queue } from 'bullmq';
import IORedis from 'ioredis-mock';

export function createMockQueue(name: string) {
  const connection = new IORedis();
  return new Queue(name, { connection });
}

class QueueFactory {
  private queues = new Map<QueueName, Queue>();

  get(name: QueueName): Queue {
    if (!this.queues.has(name)) {
      this.queues.set(name, this.create(name));
    }
    return this.queues.get(name)!;
  }

  private create(name: QueueName): Queue {
    return new Queue(name, {
      connection: defaultConnection,
    });
  }
}

class QueueFactory {
  private queues = new Map<QueueName, Queue>();

  get(name: QueueName): Queue {
    // ... 创建逻辑
  }

  async closeAll(): Promise<void> {
    const closingPromises = Array.from(this.queues.values()).map(
      queue => queue.close()
    );
    await Promise.all(closingPromises);
    this.queues.clear();
  }
}

process.on('SIGTERM', async () => {
  await queueFactory.closeAll();
  process.exit(0);
});

export function createQueue<T = any>(
  name: QueueName,
  options?: Partial<QueueOptions>
): Queue<T> {
  return new Queue<T>(name, {
    connection: defaultConnection,
    defaultJobOptions: {
      attempts: 3,
      backoff: {
        type: 'exponential',
        delay: 1000,
      },
    },
    ...options,  // 调用方可以覆盖默认配置
  });
}

// 使用默认配置
const emailQueue = createQueue('email');

// 覆盖部分配置
const imageQueue = createQueue('image', {
  defaultJobOptions: {
    attempts: 5,  // 图片处理可以多重试几次
  },
});