RabbitMQ. publishing message in loop

Question

I'm using amqplib js lib. I have publisher (it publish messages in loop) and few workers. If I publish 1000000 messages I see how my messages firstly sending to rabbit, after that I'm receiving ack, and only after this message start consuming in worker/workers.

As I understand, while I sending messages, rabbit not able to send ack to publisher. Am i right? How can I solve this?

I have main file:

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
};

simple publisher:

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

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

and simple worker:

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

runWorker();

result of work: I don't know if it's important or not, but i say. I use this rabbit cluster, and in it I turned on one policy:

also I thought, that some propblem in channel and I added one more test:

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

  let channel = await core.runInit();

  await core.runWorker(channel);

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

but result was the same:

score 0 · Answer 1 · answered Jan 14 '20 at 14:45

I do not know if I am right or not, but I solved this problem)

previously i used one connection and one channel for consumer and for publisher. but not, I open new channel for each new publish message.

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

as I saw in Rabbit Management, in Channel Tab were only one channel, even when I opened five publishers and one worker.

full code:

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
};

and now, I want to hear your opinion about this

RabbitMQ. publishing message in loop

1 Answers1