rabbitMQ ConfirmListener

消息消费者
操作步骤：
1. 创建连接工厂ConnectionFactory
2. 获取连接Connection
3. 通过连接获取通信通道Channel
4. 声明交换机Exchange：交换机类型分为四类：

        Fanout Exchange: 将消息分发到所有的绑定队列，无routingkey的概念

        Headers Exchange ：通过添加属性key-value匹配

        Direct Exchange:按照routingkey分发到指定队列

        Topic Exchange:多关键字匹配

5. 声明队列Queue

6. 将队列和交换机绑定

7. 创建消费者

8. 执行消息的消费

package org.lkl.mq.rabbitmq.test;  
  
import java.io.IOException;  
import java.util.concurrent.TimeUnit;  
import java.util.concurrent.TimeoutException;  
  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.ConnectionFactory;  
import com.rabbitmq.client.ConsumerCancelledException;  
import com.rabbitmq.client.QueueingConsumer;  
import com.rabbitmq.client.QueueingConsumer.Delivery;  
import com.rabbitmq.client.ShutdownSignalException;  
  
/**  
 * 客户端01  
 *   
 * @author liaokailin  
 * @version $Id: Receive01.java, v 0.1 2015年11月01日 下午3:47:58 liaokailin Exp $  
 */  
public class Receive01 {  
    public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException,  
                                          ConsumerCancelledException, InterruptedException {  
        ConnectionFactory facotry = new ConnectionFactory();  
        facotry.setUsername("test");  
        facotry.setPassword("test");  
        facotry.setVirtualHost("test");  
        facotry.setHost("localhost");  
  
        Connection conn = facotry.newConnection(); //获取一个链接  
        //通过Channel进行通信  
        Channel channel = conn.createChannel();  
        int prefetchCount = 1;  
        channel.basicQos(prefetchCount); //保证公平分发  
  
        boolean durable = true;  
        //声明交换机  
        channel.exchangeDeclare(Send.EXCHANGE_NAME, "direct", durable); //按照routingKey过滤  
        //声明队列  
        String queueName = channel.queueDeclare("queue-01", true, true, false, null).getQueue();  
        //将队列和交换机绑定  
        String routingKey = "lkl-0";  
        //队列可以多次绑定，绑定不同的交换机或者路由key  
        channel.queueBind(queueName, Send.EXCHANGE_NAME, routingKey);  
  
        //创建消费者  
        QueueingConsumer consumer = new QueueingConsumer(channel);  
          
        //将消费者和队列关联  
        channel.basicConsume(queueName, false, consumer); // 设置为false表面手动确认消息消费  
  
        //获取消息  
  
        System.out.println(" Wait message ....");  
        while (true) {  
            Delivery delivery = consumer.nextDelivery();  
            String msg = new String(delivery.getBody());  
            String key = delivery.getEnvelope().getRoutingKey();  
  
            System.out.println("  Received '" + key + "':'" + msg + "'");  
            System.out.println(" Handle message");  
            TimeUnit.SECONDS.sleep(3); //mock handle message  
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); //确定该消息已成功消费  
        }  
  
    }  
}  

消息生产者
操作步骤：
1. 创建连接工厂ConnectionFactory
2. 获取连接Connection
3. 通过连接获取通信通道Channel
4. 发送消息

package org.lkl.mq.rabbitmq.test;  
  
import java.io.IOException;  
import java.util.concurrent.TimeoutException;  
  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.ConfirmListener;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.ConnectionFactory;  
import com.rabbitmq.client.MessageProperties;  
  
/**  
 * 消息publish  
 *   
 * @author liaokailin  
 * @version $Id: Send.java, v 0.1 2015年10月22日 下午3:48:09 liaokailin Exp $  
 */  
public class Send {  
    public final static String EXCHANGE_NAME = "test-exchange";  
  
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {  
        /**  
         * 配置amqp broker 连接信息  
         */  
        ConnectionFactory facotry = new ConnectionFactory();  
        facotry.setUsername("test");  
        facotry.setPassword("test");  
        facotry.setVirtualHost("test");  
        facotry.setHost("localhost");  
  
        Connection conn = facotry.newConnection(); //获取一个链接  
        //通过Channel进行通信  
        Channel channel = conn.createChannel();  
  
        // channel.exchangeDeclare(Send.EXCHANGE_NAME, "direct", true); //如果消费者已创建，这里可不声明  
        channel.confirmSelect(); //Enables publisher acknowledgements on this channel  
        channel.addConfirmListener(new ConfirmListener() {  
  
            @Override  
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {  
                System.out.println("[handleNack] :" + deliveryTag + "," + multiple);  
  
            }  
  
            @Override  
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {  
                System.out.println("[handleAck] :" + deliveryTag + "," + multiple);  
            }  
        });  
  
        String message = "lkl-";  
        //消息持久化 MessageProperties.PERSISTENT_TEXT_PLAIN  
        //发送多条信息，每条消息对应routekey都不一致  
        for (int i = 0; i < 10; i++) {  
            channel.basicPublish(EXCHANGE_NAME, message + (i % 2), MessageProperties.PERSISTENT_TEXT_PLAIN,  
                (message + i).getBytes());  
            System.out.println("[send] msg " + (message + i) + " of routingKey is " + (message + (i % 2)));  
        }  
  
    }  
}  

在设置消息被消费的回调前需显示调用

引用

channel.confirmSelect() 

否则回调函数无法调用

先执行消费者，消费者会轮询是否有消息的到来，在web控制也可以观察哦～～，再启动生产者发送消息。

rabbitmq 为每一个channel维护了一个delivery tag的计数器，这里采用正向自增，新消息投递时自增，当消息响应时自减

参考：http://blog.csdn.net/liaokailin/article/details/49558605

     http://blog.csdn.net/stonexmx/article/details/51885745