0

我从 RabbitMq 服务器收到此错误

服务器关闭的频道：406（PRECONDITION-FAILED），带有消息“PRECONDITION_FAILED - 未知的交付标签 80”

发生这种情况是因为在消费者任务期间连接丢失，最后，当消息被确认/确认时，我收到此错误，因为我无法在与我从中获取消息的通道不同的通道上确认消息。

这是 RabbitMq 连接的代码

async connect({ prefetch = 1, queueName }) {
    this.queueName = queueName;
    console.log(`[AMQP][${this.queueName}] | connecting`);
    return queue
        .connect(this.config.rabbitmq.connstring)
        .then(conn => {
            conn.once('error', err => {
                this.channel = null;
                if (err.message !== 'Connection closing') {
                    console.error(
                        `[AMQP][${this.queueName}] (evt:error) | ${err.message}`,
                    );
                }
            });

            conn.once('close', () => {
                this.channel = null;
                console.error(
                    `[AMQP][${this.queueName}] (evt:close) | reconnecting`,
                );
                this.connect({ prefetch, queueName: this.queueName });
            });
            return conn.createChannel();
        })
        .then(ch => {
            console.log(`[AMQP-channel][${this.queueName}] created`);
            ch.on('error', err => {
                console.error(
                    `[AMQP-ch][${this.queueName}] (evt:error) | ${err.message}`,
                );
            });
            ch.on('close', () => {
                console.error(`[AMQP-ch][${this.queueName}] (evt:close)`);
            });
            this.channel = ch;
            return this.channel;
        })
        .then(ch => {
            return this.channel.prefetch(prefetch);
        })
        .then(ch => {
            return this.channel.assertQueue(this.queueName);
        })
        .then(async ch => {
            while (this.buffer.length > 0) {
                const request = this.buffer.pop();
                await request();
            }
            return this.channel;
        })
        .catch(error => {
            console.error(error);
            console.log(`[AMQP][${this.queueName}] reconnecting in 1s`);
            return this._delay(1000).then(() =>
                this.connect({ prefetch, queueName: this.queueName }),
            );
        });
}

async ack(msg) {
    try {
        if (this.channel) {
            console.log(`[AMQP][${this.queueName}] ack`);
            await this.channel.ack(msg);
        } else {
            console.log(`[AMQP][${this.queueName}] ack (buffer)`);
            this.buffer.push(() => {
                this.ack(msg);
            });
        }
    } catch (e) {
        console.error(`[AMQ][${this.queueName}] ack error: ${e.message}`);
    }
}

如您所见，在建立连接后，创建了一个通道，并且在我遇到连接问题后，通道设置为 NULL，1 秒后连接重试，重新创建一个新通道。

为了管理离线期间，我使用了一个缓冲区，该缓冲区收集在通道为 NULL 时发送的所有确认消息，并且在重新建立连接后我卸载缓冲区。

所以基本上我必须找到一种方法在连接丢失或通道因watherver原因关闭后发送ACK。

谢谢你的帮助

4

2 回答 2

1

一旦通道关闭（无论是什么原因），您就无法确认消息。代理将自动将相同的消息重新传递给另一个消费者。

这在 RabbitMQ消息确认部分中有详细记录。

当消费者失败或失去连接时：自动重新排队

当使用手动确认时，任何未确认的传递（消息）都会在传递发生的通道（或连接）关闭时自动重新排队。这包括客户端的 TCP 连接丢失、消费者应用程序（进程）故障和通道级协议异常（如下所述）。

...

由于这种行为，消费者必须准备好处理重新交付，否则在实施时要牢记幂等性。Redeliveries 将有一个特殊的布尔属性，redeliver，由 RabbitMQ 设置为 true。对于首次交付，它将设置为 false。请注意，消费者可以接收以前传递给另一个消费者的消息。

正如文档所建议的，您需要通过实现消息幂等性设计模式在消费者端处理此类问题。换句话说，您的架构应该准备好处理由于错误导致的消息重新传递。

或者，您可以禁用消息确认并获得“一次交付”类型的模式。这意味着如果出现错误，您将不得不处理消息丢失。

有关此事的进一步阅读： https ://bravenewgeek.com/you-cannot-have-exactly-once-delivery/

Kafka 引入新语义后的后续行动： https ://bravenewgeek.com/you-cannot-have-exactly-once-delivery-redux/

于 2021-07-06T09:13:11.457 回答
0

如果连接由于某种原因被丢弃或中断，则无法发送 ACK，因为连接发生在套接字级别，一旦关闭，就无法使用同一个套接字重新创建它。

当连接断开时，消息保持非 ACK，因此另一个侦听器可以处理它，或者当它再次连接时，断开连接的侦听器将再次处理它。

在我看来，您正在尝试解决一个不是 RabbitMQ 而是由基础的套接字实现给出的问题。

您可以通过避免管理消息缓冲区并利用 RabbitMQ 的特性来解决此问题，一旦您的侦听器再次连接，它将重新呈现最后一条未处理的消息。

于 2021-07-06T09:08:37.790 回答