如何确保在关闭连接之前成功发布最后一条 AMQP 消息？答案

【问题标题】：How to make sure last AMQP message is published successfully before closing connection?如何确保在关闭连接之前成功发布最后一条 AMQP 消息？
【发布时间】：2017-01-17 08:48:53
【问题描述】：

我有多个进程作为一个系统协同工作。其中一个进程充当主进程。当系统关闭时，每个进程都需要向主进程发送通知（通过 RabbitMQ）然后退出。该程序是用 C++ 编写的，我使用的是AMQPCPP library。

问题是有时通知没有成功发布。我怀疑退出太早是问题的原因，因为 AMQPCPP 库在关闭其连接之前没有机会发送消息。

AMQPCPP 的文档说：

发布的消息通常不会被服务器确认，并且 RabbitMQ 不会返回报告来通知您消息是否成功发布。因此，发布方法不会返回 Deferred 对象。

只要没有通过 Channel::onError() 方法报告错误，您就可以放心地假定您的消息已送达。

当您发布许多消息时，这当然会成为一个问题。如果您在中途遇到错误，则无法确定有多少消息发送到代理以及应该重新发布多少消息。如果这很重要，您可以将发布命令包装在事务中。在这种情况下，如果发生错误，事务会被 RabbitMQ 自动回滚，并且不会真正发布任何消息。

如果没有 RabbitMQ 服务器的确认，很难确定何时可以安全退出进程。此外，使用事务处理通知听起来有点矫枉过正。

谁能提出一个简单的解决方案来优雅地关闭而不会丢失最后的通知？

【问题讨论】：

标签： c++ rabbitmq amqp

【解决方案1】：

原来我可以在关闭频道时设置回调。这样我就可以在所有通道成功关闭时安全地关闭连接。我不完全确定这个过程是否能确保所有传出消息都被真正发布。不过从测试结果来看，问题似乎解决了。

class MyClass
{
    ...
    AMQP::TcpConnection m_tcpConnection;
    AMQP::TcpChannel m_channelA;
    AMQP::TcpChannel m_channelB;
    ...
};

void MyClass::stop(void)
{
    sendTerminateNotification();

    int remainChannel = 2;
    auto closeConnection = [&]() {
        --remainChannel;
        if (remainChannel == 0) {
            // close connection when all channels are closed.
            m_tcpConnection.close();
            ev::get_default_loop().break_loop();
        }
    };

    auto closeChannel = [&](AMQP::TcpChannel & channel) {
        channel.close()
            .onSuccess([&](void) { closeConnection(); })
            .onError([&](const char * msg)
                {
                    std::cout << "cannot close channel: "
                            << msg << std::endl;
                    // close the connection anyway
                    closeConnection();
                }
            );

    closeChannel(m_channelA);
    closeChannel(m_channelB);
}

【讨论】：