/**
 * @param exchange the name of the exchange  > 交换机名称
 * @param type the exchange type  > 交换机类型
 * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)  
   > 持久化标志
 * @param autoDelete true if the server should delete the exchange when it is no longer in use
 * @param internal true if the exchange is internal, i.e. can't be directly
 * published to by a client.
 * @param arguments other properties (construction arguments) for the exchange
 * @return a declaration-confirm method to indicate the exchange was successfully declared
 * @throws java.io.IOException if an error is encountered
 */
Exchange.DeclareOk exchangeDeclare(String exchange,
	BuiltinExchangeType type,
	boolean durable,
	boolean autoDelete,
	boolean internal,
	Map<String, Object> arguments) throws IOException;
	
	
void exchangeDeclareNoWait(String exchange, BuiltinExchangeType type, boolean durable,
        boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException;
Exchange.DeclareOk exchangeDeclarePassive(String name) throws IOException;

/**
 * Declare a queue
 * @see com.rabbitmq.client.AMQP.Queue.Declare
 * @see com.rabbitmq.client.AMQP.Queue.DeclareOk
 * @param queue the name of the queue  > 队列的名称
 * @param durable true if we are declaring a durable queue (the queue will survive a server restart)  >持久化标志
 * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
   > 排他队列，如果一个队列被声明为排他队列，该队列仅对首次申明它的连接可见，并在连接断开时自动删除。这里需要注意三点：
   1.排他队列是基于连接可见的，同一连接的不同信道是可以同时访问同一连接创建的排他队列；
   2.“首次”，如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同；
   3.即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除的，这种队列适用于一个客户端发送读取消息的应用场景。
 * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
   > 自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。
 * @param arguments other properties (construction arguments) for the queue
 * @return a declaration-confirm method to indicate the queue was successfully declared
 * @throws java.io.IOException if an error is encountered
 */
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
							 Map<String, Object> arguments) throws IOException;

/**
 * @param exchange the exchange to publish the message to  > 交换机名称
 * @param routingKey the routing key  > 路由键
 * @param mandatory true if the 'mandatory' flag is to be set
   > 当mandatory标志位设置为true时，如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，
   那么会调用basic.return方法将消息返回给生产者(Basic.Return + Content-Header + Content-Body)；当mandatory设置为false时，出现上述情形broker会直接将消息扔掉。
 * @param immediate true if the 'immediate' flag is to be
 * set. Note that the RabbitMQ server does not support this flag.
   > 当immediate标志位设置为true时，如果exchange在将消息路由到queue(s)时发现对于的queue上么有消费者，
   那么这条消息不会放入队列中。当与消息routeKey关联的所有queue(一个或者多个)都没有消费者时，该消息会通过basic.return方法返还给生产者。
   !! 在RabbitMQ3.0以后的版本里，去掉了immediate参数的支持
 * @param props other properties for the message - routing headers etc  消息属性字段，比如消息头部信息等等
 * @param body the message body  消息主体部分
 * @throws java.io.IOException if an error is encountered
 */
void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
		throws IOException;

body可用com.fasterxml.jackson.databind.ObjectMapper将String转为byte数组。

public BasicProperties(
		String contentType, // 消息类型如：text/plain
		String contentEncoding, // 编码
		Map<String,Object> headers,
		Integer deliveryMode, // 1:nonpersistent不持久化 2:persistent持久化
		Integer priority, // 优先级
		String correlationId,
		String replyTo, // 反馈队列
		String expiration, // expiration到期时间
		String messageId,
		Date timestamp,
		String type,
		String userId,
		String appId,
		String clusterId)

public static final BasicProperties PERSISTENT_TEXT_PLAIN =
        new BasicProperties("text/plain",
                            null,
                            null,
                            2,
                            0, null, null, null,
                            null, null, null, null,
                            null, null);

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.deliveryMode(2);
AMQP.BasicProperties properties = builder.build();

try {
	channel.txSelect();
	...
	channel.basicPublish(exchange, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
	...
	channel.txCommit();
catch (Exception e) {
	...
	channel.txRollback();
}

// producer的手工确认模式
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(ip, port);
connectionFactory.setUsername(userName);
connectionFactory.setPassword(password);
connectionFactory.setPublisherConfirms(true); // enable confirm mode
...

@Bean
/** 因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最后一次设置 */
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public RabbitTemplate rabbitTemplate() {
	RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
	return template;
}


rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
	@Override
	public void confirm(CorrelationData correlationData, boolean ack, String cause) {
		if (!ack) {
			throw new RuntimeException("send error " + cause);
		}
	}
});
// rabbitTemplate.setReturnCallback()

可以通过为channel信道设置ReturnListener监听器来实现

channel.addReturnListener(new ReturnListener() {
	@Override
	public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties basicProperties, byte[] body) throws IOException {
		String message = new String(body);
		System.out.println("Basic.return返回的结果是: " + message);
	}
});

Removal of “immediate” flag 
What changed? We removed support for the rarely-used “immediate” flag on AMQP’s basic.publish. 
Why on earth did you do that? Support for “immediate” made many parts of the codebase more complex, particularly around mirrored queues. It also stood in the way of our being able to deliver substantial performance improvements in mirrored queues. 
What do I need to do? If you just want to be able to publish messages that will be dropped if they are not consumed immediately, you can publish to a queue with a TTL of 0. 
If you also need your publisher to be able to determine that this has happened, you can also use the DLX feature to route such messages to another queue, from which the publisher can consume them. 

大意: immediate标记会影响镜像队列性能，增加代码复杂性，并建议采用“TTL”和“DLX”等方式替代。

rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]

-p Vhost： 可选参数，针对指定vhost下的queue进行设置
Name: policy的名称
Pattern: queue的匹配模式(正则表达式)
Definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode
    ha-mode:指明镜像队列的模式，有效值为 all/exactly/nodes
        all：表示在集群中所有的节点上进行镜像
        exactly：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定
        nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定
    ha-params：ha-mode模式需要用到的参数
    ha-sync-mode：进行队列中消息的同步方式，有效值为automatic和manual
priority：可选参数，policy的优先级

例如，对队列名称以“queue_”开头的所有队列进行镜像，并在集群的两个节点上完成进行，policy的设置命令为：
rabbitmqctl set_policy --priority 0 --apply-to queues mirror_queue "^queue_" '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, false, consumer);
while(true){
	QueueingConsumer.Delivery delivery = consumer.nextDelivery();
	String msg = new String(delivery.getBody());
	// do something with msg. 
	channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
	
	// requeue，true重新进入队列  
    //channel.basicNack(envelope.getDeliveryTag(), false, true); 
	// requeue，true重新进入队列,与basicNack差异缺少multiple参数  
    //channel.basicReject(envelope.getDeliveryTag(), true);
	// 重新发送到队列中
	//channel.basicRecover();
}

使用spring-rabbit时，用Listener(实现接口ChannelAwareMessageListener)接受消息设置AcknowledgeMode

@Bean
public SimpleMessageListenerContainer testListenerContainer() {
	SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
	container.setConnectionFactory(configBase.connectionFactory());
	container.setQueues(this.testQueue());
	container.setExposeListenerChannel(true);
	container.setMaxConcurrentConsumers(1);
	container.setConcurrentConsumers(1);
	container.setMessageListener(myTestMQListener);
	container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
	return container;
}

<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
    <rabbit:listener queues="queue_xxx" ref="MqConsumer"/>
    <rabbit:listener queues="queue_xxx" ref="MqConsumer2"/>
</rabbit:listener-container>

spring boot 实现：

@RabbitHandler
@RabbitListener(queues = {"${rabbitmq.bill_queueName}"})
public void process(@Payload Message msg, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception {
	...
}

@RabbitListener(containerFactory = "rabbitListenerContainerFactory", bindings = @QueueBinding(
        value = @Queue(value = "${mq.config.queue}", durable = "true"),
        exchange = @Exchange(value = "${mq.config.exchange}", type = ExchangeTypes.TOPIC),
        key = "${mq.config.key}"), admin = "rabbitAdmin")