RabbitMQ系列（九）：TX与confirm

从RabbitMQ系列（三）：work queue我们学习到了使用ack来确保message消息从queue到consumer阶段的不会被丢失，学习到使用durable=true来设置消息持久化，以保证消息在queue中还未被消费时，RabbitMQ Server重启消息不会丢失，但这并不是完美的解决方案，因为message在存储至disk的这个过程中仍然需要一小段时间来完成这个任务，倘若在这段时间内Server宕机，那么消息也会丢失。

在使用RabbitMQ的时候，我们可以通过消息持久化操作来解决因为服务器的异常奔溃导致的消息丢失，除此之外我们还会遇到一个问题，当消息的发布者在将消息发送出去之后，消息到底有没有正确到达broker代理服务器呢？如果不进行特殊配置的话，默认情况下发布操作是不会返回任何信息给生产者的，也就是默认情况下我们的生产者是不知道消息有没有正确到达broker的，如果在消息到达broker之前已经丢失的话，持久化操作也解决不了这个问题，因为消息根本就没到达代理服务器，怎么进行持久化，要解决这个问题

RabbitMQ为我们提供了两种方式：

通过AMQP事务机制实现，这也是AMQP协议层面提供的解决方案；

通过将channel设置成confirm模式来实现；

TX

对于一个channel，我们如果要用事务机制保证数据的一致性，则需要将channel设置成transaction模式，函数Channel.Tx()，Channel.TxCommit()用于提交事务，Channel.TxRollback()用于回滚事务。在事务开启时候，我们便可以发布消息给broker代理服务器，如果TxCommit()提交成功，则message一定到达了broker；如果TxCommit()执行时返回一个错误，即broker有异常。这时候就可以用TxRollback()回滚事务。

在接下里的例子中，我们会将本节内容与RabbitMQ系列（三）：work queue中的消息持久化和消息ack确认机制共同使用，以保证消息在通过RabbitMQ的过程中不会丢失。

sender.go

声明一个queue

将channel置为transaction模式，发布消息并提交tx

函数bodyFrom()

receiver.go

声明同样的queue（略）

接收消息

由于事务机制比没有事务多了四个步骤：

（1）client发送Tx.Select

（2）broker发送Tx.Select-Ok(之后publish)

（3）client发送Tx.Commit

（4）broker发送Tx.Commit-Ok

所以这里是有性能损耗的，那么有没有更好的方法既能保障producer知道消息已经正确送到，又能基本上不带来性能上的损失呢？从AMQP协议的层面看是没有更好的方法，但是RabbitMQ提供了一个更好的方案，即将channel信道设置成confirm模式。

confirm

producer端与confirm的实现原理

生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者（包含消息的唯一ID）,这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会将消息写入磁盘之后发出，broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息。

在channel 被设置成 confirm 模式之后，所有被 publish 的后续消息都将被 confirm（即 ack） 或者被nack一次。但是没有对消息被 confirm 的快慢做任何保证，并且同一条消息不会既被 confirm又被nack。

开启confirm模式

以confirm模式发送消息

其他辅助函数confirmOne()

当Channel.Confirm(noWait bool)参数设置为false时，broker会返回一个confirm.ok表示同意发送者将当前channel信道设置为confirm模式。

其他代码和transaction模式类似，只是没有Channel.TxCommit()和Channel.TxRollback()。

编程模式

对于固定消息体大小和线程数，如果消息持久化，生产者confirm(或者采用事务机制)，消费者ack那么对性能有很大的影响.

消息持久化的优化没有太好方法，用更好的物理存储（SAS, SSD, RAID卡）总会带来改善。生产者confirm这一环节的优化则主要在于客户端程序的优化之上。归纳起来，客户端实现生产者confirm有三种编程方式：

普通confirm模式：每发送一条消息后，调用waitForConfirms()方法，等待服务器端confirm。实际上是一种串行confirm了。

批量confirm模式：每发送一批消息后，调用waitForConfirms()方法，等待服务器端confirm。

异步confirm模式：提供一个回调方法，服务端confirm了一条或者多条消息后Client端会回调这个方法。

从编程实现的复杂度上来看：

第1种

普通confirm模式最简单，publish一条消息后，等待服务器端confirm,如果服务端返回false或者超时时间内未返回，客户端进行消息重传。

第二种

批量confirm模式稍微复杂一点，客户端程序需要定期（每隔多少秒）或者定量（达到多少条）或者两则结合起来publish消息，然后等待服务器端confirm, 相比普通confirm模式，批量极大提升confirm效率，但是问题在于一旦出现confirm返回false或者超时的情况时，客户端需要将这一批次的消息全部重发，这会带来明显的重复消息数量，并且，当消息经常丢失时，批量confirm性能应该是不升反降的。

第三种

异步confirm模式的编程实现最复杂，Channel对象提供的ConfirmListener()回调方法只包含deliveryTag（当前Chanel发出的消息序号），我们需要自己为每一个Channel维护一个unconfirm的消息序号集合，每publish一条数据，集合中元素加1，每回调一次handleAck方法，unconfirm集合删掉相应的一条（multiple=false）或多条（multiple=true）记录。从程序运行效率上看，这个unconfirm集合最好采用有序集合SortedSet存储结构。实际上，SDK中的waitForConfirms()方法也是通过SortedSet维护消息序号的。