2

我正在使用amqplib js 库。我有发布者（它循环发布消息）和少数工作人员。如果我发布 1000000 条消息，我会看到我的消息如何首先发送给 rabbit，然后我收到ack，并且只有在这条消息开始在 worker/workers 中消费之后。

据我了解，当我发送消息时，兔子无法向发布者发送ack。我对吗？我该如何解决这个问题？

我有主文件：

let amqp = require('amqplib');

const EXCHANGE = 'simple_exchange',
  EXCHANGE_TYPE = 'direct',
  QUEUE = 'simple_queue',
  ROUTING_KEY = 'simple_routing_key';

const defaultPublishCount = 10000;

let runInit = async() => {
  let connection = await amqp.connect('amqp://localhost');
  let channel = await connection.createChannel();

  let commonOptions = {
    durable: false
  };

  await channel.assertExchange(EXCHANGE, EXCHANGE_TYPE, commonOptions);
  await channel.assertQueue(QUEUE, commonOptions);
  await channel.bindQueue(QUEUE, EXCHANGE, ROUTING_KEY, commonOptions);

  return channel;
};

let runPublisher = async(count, userChannel) => {
  const channel = userChannel || await runInit();

  let d1 = (new Date).toISOString().slice(11, 23).replace('T', ' ');
  const publishCount = count || defaultPublishCount;

  let index = 1;
  while (index <= publishCount) {
    let msg = {
      id: index,
      time: (new Date).toISOString().slice(11, 23).replace('T', ' ')
    };
    channel.publish(EXCHANGE, ROUTING_KEY, new Buffer(JSON.stringify(msg)), {
      noAck: true
    });
    console.log(`Message sent: ${JSON.stringify(msg)}`);
    index++;
  }
  let d2 = (new Date).toISOString().slice(11, 23).replace('T', ' ');


  console.log(`\nPublish started  at: ${d1}`);
  console.log(`Publish finished at: ${d2}\n\n`);
};

let runWorker = async(userChannel) => {
  const channel = userChannel || await runInit();

  console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", QUEUE);
  channel.consume(QUEUE, (msg) => {
    console.log(` [x] Received: '${msg.content}' at ${(new Date).toISOString().slice(11, 23).replace('T', ' ')}`);
    channel.ack(msg);
  });
};

module.exports = {
  runInit,
  runPublisher,
  runWorker
};

简单的发布者：

let { runPublisher } = require('./amqp_core.js');

let count = (process.argv.slice(2, 3)[0]) * 1;
runPublisher(count);

和简单的工人：

let { runWorker } = require('./amqp_core.js');

runWorker();

工作成果： 我不知道它是否重要，但我说。我使用这个 rabbit cluster，并在其中启用了一项策略： 10 000 条消息1 000 000 条消息

我还想，频道中有一些问题，我又添加了一个测试：

(async() => {
  let core = require('./amqp_core.js');

  let channel = await core.runInit();

  await core.runWorker(channel);

  core.runPublisher(25, channel);
})();

但结果是一样的： img2

4

1 回答 1

0

我不知道我是否正确，但我解决了这个问题）

在此处输入图像描述 以前我为消费者和发布者使用了一个连接和一个通道。但不是，我为每个新的发布消息打开新频道。

let channel = await connection.createChannel();
await channel.publish(EXCHANGE, ROUTING_KEY, new Buffer(JSON.stringify(msg)), {noAck: true});
await channel.close();

正如我在 Rabbit Management 中看到的，在 Channel Tab 中只有一个频道，即使我开了五个发布者和一个工人。

完整代码： 

let amqp = require('amqplib');

const EXCHANGE = 'simple_exchange',
  EXCHANGE_TYPE = 'direct',
  QUEUE = 'simple_queue',
  ROUTING_KEY = 'simple_routing_key';

const defaultPublishCount = 10000;

let runInit = async() => {
  let connection = await amqp.connect('amqp://localhost');
  let channel = await connection.createChannel();

  let commonOptions = {
    durable: false
  };

  await channel.assertExchange(EXCHANGE, EXCHANGE_TYPE, commonOptions);
  await channel.assertQueue(QUEUE, commonOptions);
  await channel.bindQueue(QUEUE, EXCHANGE, ROUTING_KEY, commonOptions);
  await channel.close();

  return connection;
};

let runPublisher = async(count) => {
  const connection = await runInit();

  let d1 = (new Date).toISOString().slice(11, 23).replace('T', ' ');
  const publishCount = count || defaultPublishCount;

  let index = 1;
  while (index <= publishCount) {
    let msg = {
      id: index,
      time: (new Date).toISOString().slice(11, 23).replace('T', ' ')
    };
    let channel = await connection.createChannel();
    await channel.publish(EXCHANGE, ROUTING_KEY, new Buffer(JSON.stringify(msg)), {
      noAck: true
    });
    await channel.close();
    console.log(`Message sent: ${JSON.stringify(msg)}`);
    index++;
  }
  let d2 = (new Date).toISOString().slice(11, 23).replace('T', ' ');


  console.log(`\nPublish started  at: ${d1}`);
  console.log(`Publish finished at: ${d2}\n\n`);
};

let runWorker = async() => {
  const connection = await runInit();
  let channel = await connection.createChannel();

  console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", QUEUE);
  channel.consume(QUEUE, (msg) => {
    console.log(` [x] Received: '${msg.content}' at ${(new Date).toISOString().slice(11, 23).replace('T', ' ')}`);
    channel.ack(msg);
  });
};

module.exports = {
  runInit,
  runPublisher,
  runWorker
};

现在，我想听听你对此的看法

于 2020-01-14T14:45:33.363 回答