【问题标题】：Await for proper connection to rabbit in nodejs在nodejs中等待与rabbit的正确连接
【发布时间】：2019-02-27 21:05:33
【问题描述】：

我尝试为 amqplib/callback_api 编写简单的事件发射器包装器。当兔子不可用或断开连接时，我无法处理情况。我有方法getConnect，它返回 Promise，它在建立连接时解决。但是如果连接被拒绝，Promise 显然会拒绝。如何在连接未建立时强制此方法重新连接

/**
     * Async method getConnect for connection
     * @returns {Promise<*>}
     */
    getConnect = async () => {
        return new Promise((resolve, reject) => {
            amqp.connect(this.config.url, async function(err, conn) {
                    if (err) {
                        reject(err);
                    }
                    resolve(conn);
            })
        })
    };

完整代码在这里https://github.com/kimonniez/rabbitEE

也许，我已经很困了，但我完全糊涂了 :) 提前谢谢！

【问题讨论】：

标签： javascript node.js rabbitmq node-amqplib

【解决方案1】：

将`Promise` 包裹在`Observable` 中

Promise 不是为处理 “重试” 逻辑而构建的。如果您想这样做，您应该使用rxjs library 查看Observables。这将允许您在捕获错误时使用任意时间间隔重试。

const { from, interval, of } = rxjs;
const { catchError, mergeMap, tap, skipWhile, take } = rxjs.operators;

const THRESHOLD = 3;
const RETRY_INTERVAL = 1000;

// Equivalent to 'amqp.connect'
const functionThatThrows = number =>
  number < THRESHOLD
    ? Promise.reject(new Error("ERROR"))
    : Promise.resolve("OK");

// Equivalent to `getConnect`
const getConnect = () =>
  interval(RETRY_INTERVAL)
    .pipe(
      mergeMap(x => from(functionThatThrows(x)).pipe(catchError(e => of(e)))),
      skipWhile(x => {
        const isError = x instanceof Error;
        if (isError) console.log('Found error. Retrying...');
        return isError;
      }),
      take(1)
    ).toPromise();

// Resolve only if the inner Promise is resolved
getConnect().then(console.log);

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.4.0/rxjs.umd.min.js"&gt;&lt;/script&gt;

说明

使用interval 或1000 创建一个源。这意味着它将每秒重试一次
打电话给你的amqp.connect，在我的例子中相当于functionThatThrows
使用catchError 运算符捕获错误并返回
当返回的对象出错时跳过。这将允许您仅在您的 Promise 已解决且未被拒绝时解决
使用take(1) 获取第一个解析结果
使用toPromise 实用函数将您的observable转换为promise
调用您的函数并附加then，就像使用标准Promise 一样

【讨论】：

非常感谢！实际上我需要为这个新逻辑修改我的代码，但无论如何现在它看起来很棒！

【解决方案2】：

如果您只想继续尝试连接直到建立连接，您可以将getConnect 方法包装成一个新的keepConnect 方法：

keepConnect = async () => {
   while (true) {
      try {
         let conn = await getConnect()
         return conn
      } catch (e) {}
  }
}

但我认为通过更改while 条件来实现类似“尝试连接n 次”之类的东西会更好。一般来说，“while true”解决方案并不干净，并且可能执行不良，并有减慢事件循环的风险（想象一下，如果 connect 方法总是在几毫秒内返回错误）。

您还可以使用keepConnect 包装器作为想法，在连接尝试之间实现渐进式延迟系统。

如果您想在连接丢失时重新连接，那么这与兔子（我不知道）和他的事件有关。

【讨论】：

谢谢你的回答

将Promise 包裹在Observable 中

说明

将`Promise` 包裹在`Observable` 中