Using standard AMQP, the only way to guarantee that a message isn't lost is by using transactions -- make the channel transactional, publish the message, commit. In this case, transactions are unnecessarily heavyweight and decrease throughput by a factor of 250. To remedy this, a confirmation mechanism was introduced.

To enable confirms, a client sends the confirm.select method. Depending on whether no-wait was set or not, the broker may respond with a confirm.select-ok. Once the confirm.select method is used on a channel, it is said to be in confirm mode. A transactional channel cannot be put into confirm mode and once a channel is in confirm mode, it cannot be made transactional.

Once a channel is in confirm mode, both the broker and the client count messages (counting starts at 1 on the first confirm.select). The broker then confirms messages as it handles them by sending a basic.ack on the same channel. The delivery-tag field contains the sequence number of the confirmed message. The broker may also set the multiple field in basic.ack to indicate that all messages up to and including the one with the sequence number have been handled.

In exceptional cases when the broker is unable to handle messages successfully, instead of a basic.ack, the broker will send a basic.nack. In this context, fields of the basic.nack have the same meaning as the corresponding ones in basic.ack and the requeue field should be ignored. By nack'ing one or more messages, the broker indicates that it was unable to process the messages and refuses responsibility for them; at that point, the client may choose to re-publish the messages.
在異常情況中，broker 將無法成功處理相應的訊息，此時 broker 將傳送 basic.nack 來代替 basic.ack 。在這個情形下，basic.nack 中各域值的含義與 basic.ack 中相應各域含義是相同的，同時 requeue 域的值應該被忽略。通過 nack 一或多條訊息，broker 表明自身無法對相應訊息完成處理，並拒絕為這些訊息的處理負責。在這種情況下，client 可以選擇將訊息 re-publish 。

After a channel is put into confirm mode, all subsequently published messages will be confirmed or nack'd once. No guarantees are made as to how soon a message is confirmed. No message will be both confirmed and nack'd.
在 channel 被設定成 confirm 模式之後，所有被 publish 的後續訊息都將被 confirm（即 ack） 或者被 nack 一次。但是沒有對訊息被 confirm 的快慢做任何保證，並且同一條訊息不會既被 confirm 又被 nack 。

An example in Java that publishes a large number of messages to a channel in confirm mode and waits for the acknowledgements can be found here.
一個 Java 示例展現了 publish 大量訊息到一個處於 confirm 模式的 channel 並等待獲取 acknowledgement 的情況，示例在這裡。

When will messages be confirmed?
訊息會在何時被 confirm？

The broker will confirm messages once:
broker 將在下面的情況中對訊息進行 confirm ：

• it decides a message will not be routed to queues
  (if the mandatory flag is set then the basic.return is sent first) or
  broker 發現當前訊息無法被路由到指定的 queues 中（如果設定了 mandatory 屬性，則 broker 會先發送basic.return）
• a transient message has reached all its queues (and mirrors) or
  非持久屬性的訊息到達了其所應該到達的所有 queue 中（和映象 queue 中）
• a persistent message has reached all its queues (and mirrors) and been persisted to disk (and fsynced) or
  持久訊息到達了其所應該到達的所有 queue 中（和映象 queue 中），並被持久化到了磁碟（被 fsync）
• a persistent message has been consumed (and if necessary acknowledged) from all its queues
  持久訊息從其所在的所有 queue 中被 consume 了（如果必要則會被 acknowledge）

Notes

The broker loses persistent messages if it crashes before said messages are written to disk. Under certain conditions, this causes the broker to behave in surprising ways.
broker 會丟失持久化訊息，如果 broker 在將上述訊息寫入磁碟前異常。在一定條件下，這種情況會導致 broker 以一種奇怪的方式執行。

For instance, consider this scenario:
例如，考慮下述情景：

1. a client publishes a persistent message to a durable queue
   一個 client 將持久訊息 publish 到持久 queue 中
2. a client consumes the message from the queue (noting that the message is persistent and the queue durable), but doesn't yet ack it,
   另一個 client 從 queue 中 consume 訊息（注意：該訊息具有持久屬性，並且 queue 是持久化的），當尚未對其進行 ack 
3. the broker dies and is restarted, and
   broker 異常重啟
4. the client reconnects and starts consuming messages.
   client 重連並開始 consume 訊息

在上述情景下，client 有理由認為訊息需要被（broker）重新 deliver 。但這並非事實：重啟（有可能）會令 broker 丟失訊息。為了確保永續性，client 應該使用 confirm 機制。如果 publisher 使用的 channel 被設定為 confirm 模式，publisher 將不會收到已丟失訊息的 ack（這是因為 consumer 沒有對訊息進行 ack ，同時該訊息也未被寫入磁碟）。

下面是用rabbitmq-c實現publish訊息確認的程式碼：

/*
 *
 * gcc -o amqp_sendstring amqp_sendstring.c utils.c -I/usr/local/rabbitmq-c/include -L/usr/local/rabbitmq-c/lib -lrabbitmq
 *
 */

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/time.h>
#include <unistd.h>
#include <stdint.h>

#include <amqp_tcp_socket.h>
#include <amqp.h>
#include <amqp_framing.h>

#include "utils.h"

int main(int argc, char const *const *argv)
{
  char const *hostname;
  int port, status;
  char const *exchange;
  char const *routingkey;
  char const *messagebody;
  amqp_socket_t *socket = NULL;
  amqp_connection_state_t conn;

  if (argc < 6) {
    fprintf(stderr, "Usage: amqp_sendstring host port exchange routingkey messagebody\n");
    return 1;
  }

  hostname = argv[1];
  port = atoi(argv[2]);
  exchange = argv[3];
  routingkey = argv[4];
  messagebody = argv[5];

  conn = amqp_new_connection();

  socket = amqp_tcp_socket_new(conn);
  if (!socket) {
    die("creating TCP socket");
  }

  status = amqp_socket_open(socket, hostname, port);
  if (status) {
    die("opening TCP socket");
  }

  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
                    "Logging in");
  amqp_channel_open(conn, 1);
  die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");

  {
    amqp_basic_properties_t props;
    props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
    props.content_type = amqp_cstring_bytes("text/plain");
    props.delivery_mode = 2; /* persistent delivery mode */

    amqp_confirm_select(conn, 1);  //在通道上開啟Publish確認
    die_on_error(amqp_basic_publish(conn,
                                    1,
                                    amqp_cstring_bytes(exchange),
                                    amqp_cstring_bytes(routingkey),
                                    0,   //mandatory標誌位,訊息不能到達佇列則返回basic.return
                                    0,   //immediate標誌位,訊息不能到達消費者返回basic.return
                                    &props,
                                    amqp_cstring_bytes(messagebody)),
                 "Publishing");
  }

  {
    /* Publish訊息後需要在當前通道上監聽返回的資訊，來判斷訊息是否成功投遞
     * 這裡要息根據投遞訊息的方式來過濾判斷幾個方法
     */
    amqp_frame_t frame;
    amqp_rpc_reply_t ret;

    if (AMQP_STATUS_OK != amqp_simple_wait_frame(conn, &frame)) {
      return;
    }

    if (AMQP_FRAME_METHOD == frame.frame_type) {
      amqp_method_t method = frame.payload.method;
      fprintf(stdout, "method.id=%08X,method.name=%s\n",
        method.id, amqp_method_name(method.id));
      switch (method.id) {
        case AMQP_BASIC_ACK_METHOD:{
          /* if we've turned publisher confirms on, and we've published a message
           * here is a message being confirmed
           */
          {
            amqp_basic_ack_t *s;
            s = (amqp_basic_ack_t *) method.decoded;
            fprintf(stdout, "Ack.delivery_tag=%d\n", s->delivery_tag);
            fprintf(stdout, "Ack.multiple=%d\n", s->multiple);
          }

          break;

        case AMQP_BASIC_NACK_METHOD:
          /* if we've turned publisher confirms on, and we've published a message
           * here is a message not being confirmed
           */
          {
            amqp_basic_nack_t *s;
            s = (amqp_basic_nack_t *) method.decoded;
            fprintf(stdout, "NAck.delivery_tag=%d\n", s->delivery_tag);
            fprintf(stdout, "NAck.multiple=%d\n", s->multiple);
            fprintf(stdout, "NAck.requeue=%d\n", s->requeue);
          }

          break;

        case AMQP_BASIC_RETURN_METHOD:
          /* if a published message couldn't be routed and the mandatory flag was set
           * this is what would be returned. The message then needs to be read.
           */
          {
            amqp_message_t message;
            amqp_basic_return_t *s;
            char str[1024];
            s = (amqp_basic_return_t *) method.decoded;
            fprintf(stdout, "Return.reply_code=%d\n", s->reply_code);
            strncpy(str, s->reply_text.bytes, s->reply_text.len); str[s->reply_text.len] = 0;
            fprintf(stdout, "Return.reply_text=%s\n", str);

            ret = amqp_read_message(conn, frame.channel, &message, 0);
            if (AMQP_RESPONSE_NORMAL != ret.reply_type) {
              return;
            }
            strncpy(str, message.body.bytes, message.body.len); str[message.body.len] = 0;
            fprintf(stdout, "Return.message=%s\n", str);

            amqp_destroy_message(&message);
          }

          break;

        case AMQP_CHANNEL_CLOSE_METHOD:
          /* a channel.close method happens when a channel exception occurs, this
           * can happen by publishing to an exchange that doesn't exist for example
           *
           * In this case you would need to open another channel redeclare any queues
           * that were declared auto-delete, and restart any consumers that were attached
           * to the previous channel
           */
          return;

        case AMQP_CONNECTION_CLOSE_METHOD:
          /* a connection.close method happens when a connection exception occurs,
           * this can happen by trying to use a channel that isn't open for example.
           *
           * In this case the whole connection must be restarted.
           */
          return;

        default:
          fprintf(stderr ,"An unexpected method was received %d\n", frame.payload.method.id);
          return;
      }
    }
  }

  }

  die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
  die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
  die_on_error(amqp_destroy_connection(conn), "Ending connection");

  return 0;
}