如何解决 amqplib 的 Channel#consume 奇数签名？

Question

我正在编写一个使用 amqplib 的 Channel#consume 方法的工作者。我希望该工作人员等待作业并在它们出现在队列中时立即处理它们。

我编写了自己的模块来抽象出 ampqlib，以下是获取连接、设置队列和使用消息的相关函数：

const getConnection = function(host) {
  return amqp.connect(host);
};

const createChannel = function(conn) {
  connection = conn;
  return conn.createConfirmChannel();
};

const assertQueue = function(channel, queue) {
  return channel.assertQueue(queue);
};

const consume = Promise.method(function(channel, queue, processor) {
  processor = processor || function(msg) { if (msg) Promise.resolve(msg); };
  return channel.consume(queue, processor)
});

const setupQueue = Promise.method(function setupQueue(queue) {
  const amqp_host = 'amqp://' + ((host || process.env.AMQP_HOST) || 'localhost');

  return getConnection(amqp_host)
    .then(conn => createChannel(conn)) // -> returns a `Channel` object
    .tap(channel => assertQueue(channel, queue));
});

consumeJob: Promise.method(function consumeJob(queue) {
  return setupQueue(queue)
    .then(channel => consume(channel, queue))
  });

我的问题是 Channel#consume 的奇怪签名。来自http://www.squaremobius.net/amqp.node/channel_api.html#channel_consume：

#consume(queue, function(msg) {...}, [options, [function(err, ok) {...}]])

回调不是魔法发生的地方，消息的处理实际上应该在第二个参数中进行，这会破坏 Promise 的流程。

这是我计划使用它的方式：

return queueManager.consumeJob(queue)
  .then(msg => {
     // do some processing
  });

但它不起作用。如果队列中没有消息，则 Promise 被拒绝，然后如果消息被丢弃在队列中，则什么也不会发生。如果有消息，则只处理一条消息，然后 worker 停止，因为它从 Channel#consume 调用中退出了“处理器”函数。

我应该怎么做？我想保留 queueManager 抽象，这样我的代码更容易推理，但我不知道该怎么做......有什么指针吗？

Answer 1

正如@idbehold 所说，Promise 只能解决一次。如果您想在消息进入时对其进行处理，除了使用此功能之外别无他法。 Channel#get 只会检查一次队列然后返回；它不适用于需要工人的场景。

Answer 2

只是作为一种选择。您可以将您的应用程序呈现为一些消息（或事件）的流。这个http://highlandjs.org/#examples有一个库

你的代码应该是这样的（它不是一个完成的示例，但我希望它能说明这个想法）：

let messageStream = _((push, next) => {
  consume(queue, (msg) => {
   push(null, msg)
  })
)
// now you can operate with your stream in functional style 
message.map((msg) => msg + 'some value').each((msg) => // do something with msg)

这种方法为您提供了许多用于同步和转换的原语
http://highlandjs.org/#examples