@actbox/queue 中文文档教程

Question

ActBox Queue

该模块定义了消息代理服务的接口，以便与其余的 ActBox 服务进行通信

Setup

在开始使用该库之前，请记住，您需要安装并运行 rabbitMQ。

提示 请记住，您可以使用 Docker 毫不费力地运行 rabbitMQ Docker Hub

Configuration

这些是用于制作的可用配置选项队列。 只有 queueName 字段是必需的。

{
  queueName: "my-queue", // take in mind, you need to use the same name in producer and consumer entities as rabbitMQ docs explains

  /**
   * by default, connectionURI is "amqp://localhost" and the port is the default from rabbitMQ, it is "5672"
  **/
  connectionURI: "amqp://localhost", // it represents the end-point where rabbitMQ is allocated

  forceNACK: false // https://www.rabbitmq.com/confirms.html
}

Development

克隆队列模块后：

cd queue && npm run build

在后端和文件服务模块上，将 amqplib 替换为 @actbox/queue：

"dependencies": {
    ...
    "amqplib": "^0.6.0", (remove this dependency)
    "@actbox/queue": "file:{relativePathToQueueModule}",
    ...
  }

Usage

Work Queues

这取决于您启动消费者或生产者，您必须启动其中任何一个。

对于生产者：

const { workQueue } = require('@actbox/queue');
const { startProducer, sendToQueue } = workQueue;

await startProducer({ queueName: 'my-queue' });

const message = { name: 'actBox rules! ????' };

sendToQueue(message);

对于消费者：

const { workQueue } = require('@actbox/queue');
const { startConsumer, consumeFromQueue } = workQueue;

await startConsumer({ queueName: 'my-queue' });

consumeFromQueue().listen(async (message) => { 
  message.content; //  { name: 'actBox rules! ????' }
});

RPC

对于服务器：

const { rpc } = require('@actbox/queue');
const { startServer } = rpc;

const myQueueServer = await startServer({ queueName: 'my-queue' });

myQueueServer.listen(async (message) => {
  try {
    const response = await anyAsyncTask();

    return { response }
  } catch(err) {
    return { 
      error: { 
        message: err.message 
      }
    }
  }
});

对于客户端：

const { rpc } = require('@actbox/queue');
const { startClient } = rpc;

const myQueueClient = await startClient({ queueName: 'my-queue' });

myQueueClient.send({ data: 'some data to server' });

myQueueClient.listen((message, error) => {
  if(error) {
    // handle error as error.message
  }
  // handle response message
});

Dependencies

• amqplib
• uuid
• winston