基于Priority Queue实现Discord任务队列的技术求助（RabbitMQ相关）

核心需求 #

• 受Discord全局速率限制，同一时间只能执行一个任务
• 任务分为高、低两个优先级
• 当前任务完成后，优先执行高优先级任务（若存在）

现有实现问题 #

你当前用RabbitMQ的direct交换机+双队列的方式，同时监听高低优先级队列，但这种方式无法实现"低优先级任务执行过程中，有高优先级任务就暂停切换"的效果——因为两个队列的消费者是独立启动的，RabbitMQ会同时给两个消费者分配任务，低优先级任务一旦被消费者获取就会执行到底，没法被高优先级任务抢占。

推荐解决方案 #

方案1：单优先级队列（最简洁可靠） #

利用RabbitMQ内置的队列优先级特性，只需要一个队列，生产者发送任务时指定优先级，队列会自动优先出队高优先级任务；同时配合prefetch(1)确保同一时间只处理一个任务，完美适配你的需求。

实现代码

const amqp = require('amqplib');

// 初始化支持优先级的队列
async function setupQueue() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // 声明队列，设置最大优先级为10（0最低，10最高）
  await channel.assertQueue('discord_task_queue', {
    durable: true,
    arguments: { 'x-max-priority': 10 }
  });

  console.log('队列初始化完成');
  await channel.close();
  await connection.close();
}

// 发布任务：第二个参数指定优先级，默认低优先级
async function publishTask(content, priority = 1) {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  await channel.sendToQueue('discord_task_queue', Buffer.from(content), {
    priority,
    persistent: true // 持久化任务，防止RabbitMQ重启丢失
  });

  console.log(`已发布任务：${content} | 优先级：${priority}`);
  await channel.close();
  await connection.close();
}

// 启动消费者
async function startConsumer() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  await channel.assertQueue('discord_task_queue', {
    durable: true,
    arguments: { 'x-max-priority': 10 }
  });

  // 每次只预取1个任务，保证同一时间仅执行一个
  channel.prefetch(1);

  channel.consume('discord_task_queue', async (msg) => {
    if (!msg) return;

    try {
      const task = msg.content.toString();
      console.log(`开始执行任务：${task}`);
      // 替换为实际的Discord API调用逻辑
      await executeDiscordTask(task);
      channel.ack(msg);
      console.log(`任务完成：${task}`);
    } catch (err) {
      console.error(`任务执行失败：${msg.content.toString()} | 错误：${err}`);
      // 失败后重新入队（根据需求调整）
      channel.nack(msg, false, true);
    }
  }, { noAck: false });

  console.log('消费者已启动，等待任务...');
}

// 模拟Discord API任务执行
async function executeDiscordTask(task) {
  return new Promise(resolve => setTimeout(resolve, 5000));
}

// 测试用例（取消注释即可运行）
// setupQueue().then(() => {
//   publishTask('日常消息推送', 1);
//   publishTask('紧急通知发送', 10);
//   publishTask('数据统计更新', 1);
//   startConsumer();
// }).catch(console.error);

startConsumer().catch(console.error);

方案优势

• 完全利用RabbitMQ原生机制，无需复杂逻辑
• 自动保证高优先级任务先被处理，天然满足优先级需求
• 单队列+prefetch(1)完美适配Discord全局速率限制

方案2：双队列动态切换（兼容原有队列结构） #

如果必须保留高低两个独立队列，可以通过主动拉取+动态切换的方式实现优先级抢占：只在高优先级队列为空时，才消费低优先级队列；每次执行低任务前先检查高队列是否有新任务。

实现代码

const amqp = require('amqplib');

async function startPriorityConsumer() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // 声明原有队列
  await channel.assertQueue('high_priority_queue', { durable: true });
  await channel.assertQueue('low_priority_queue', { durable: true });

  channel.prefetch(1);

  // 消费高优先级任务
  async function consumeHigh() {
    const msg = await channel.get('high_priority_queue', { noAck: false });
    if (msg) {
      console.log(`执行高优先级任务：${msg.content.toString()}`);
      await executeDiscordTask(msg.content.toString());
      channel.ack(msg);
      return consumeHigh(); // 继续处理高队列
    }
    return consumeLow(); // 高队列为空，切换到低队列
  }

  // 消费低优先级任务，执行前先检查高队列
  async function consumeLow() {
    // 先检查是否有高优先级任务
    const highMsg = await channel.get('high_priority_queue', { noAck: false });
    if (highMsg) {
      console.log(`发现高优先级任务，切换执行：${highMsg.content.toString()}`);
      await executeDiscordTask(highMsg.content.toString());
      channel.ack(highMsg);
      return consumeHigh();
    }

    // 没有高任务，处理低任务
    const lowMsg = await channel.get('low_priority_queue', { noAck: false });
    if (lowMsg) {
      console.log(`执行低优先级任务：${lowMsg.content.toString()}`);
      await executeDiscordTask(lowMsg.content.toString());
      channel.ack(lowMsg);
      return consumeHigh(); // 完成后先检查高队列
    }

    // 两个队列都空，等待1秒后再轮询
    await new Promise(resolve => setTimeout(resolve, 1000));
    return consumeHigh();
  }

  // 启动消费循环
  consumeHigh().catch(console.error);
  console.log('优先级消费者已启动');
}

// 模拟Discord任务
async function executeDiscordTask(task) {
  return new Promise(resolve => setTimeout(resolve, 5000));
}

startPriorityConsumer().catch(console.error);

方案说明

• 通过channel.get()主动拉取任务，而非被动监听，实现队列切换逻辑
• 每次处理完任务或轮询时，优先检查高优先级队列，确保高任务能被及时处理
• 缺点是需要轮询，效率略低于单优先级队列方案

内容的提问来源于stack exchange，提问作者Joy Gupta