 public class Product {

     private static final String EXCHANGE_NAME = "exclusive.exchange";

     private static final String QUEUE_NAME = "exclusive.queue";

     private static final String ROUTING_KEY_NAME = "exclusive.routing.key";

     public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

         Connection connection = RabbitMqUtils.getConnection();

         Channel channel = connection.createChannel();

         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

         channel.queueDeclare(QUEUE_NAME, true, true, false, null);

         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_NAME);

         channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,

                 "排它队列".getBytes());

         Connection connection2 = RabbitMqUtils.getConnection();

         Channel channel2 = connection2.createChannel();

         channel2.basicConsume(QUEUE_NAME, new DefaultConsumer(channel2) {

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                 System.err.println("收到消息：" + new String(body));

             }

         });

         // 线程休眠5秒，待消息回调完成

         TimeUnit.SECONDS.sleep(5);

         RabbitMqUtils.close(connection, channel);

     }

 }

代码中18行用的是channel2，也就是connection2创建的信道，此时不满足排它队列的只允许首次创建它的连接使用，所以会抛出异常；将16行注释掉，17行的connection2替换成connection便能正常运行。

2、autoDelete：自动删除，true-自动删除，false-非自动删除。

设置为自动删除的队列后，该队列在其消费者都断开连接后，自动删除，不管队列里是否存在数据。

3、arguments：队列的其它参数。

前面几个参数很简单，我们这里只讲重点的x-dead-letter-exchange、x-dead-letter-routing-key。

 public class DeadProduct {

     private static final String EXCHANGE_NAME = "normal.exchange";

     private static final String DEAD_EXCHANGE_NAME = "dead.exchange";

     private static final String QUEUE_NAME = "normal.queue";

     private static final String DEAD_QUEUE_NAME = "dead.queue";

     private static final String ROUTING_KEY_NAME = "normal.routing.key";

     private static final String DEAD_ROUTING_KEY_NAME = "dead.routing.key";

     public static void main(String[] args) throws IOException, TimeoutException {

         Connection connection = RabbitMqUtils.getConnection();

         Channel channel = connection.createChannel();

         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

         channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

         channel.queueDeclare(QUEUE_NAME, false, false, false, getArguments());

         channel.queueDeclare(DEAD_QUEUE_NAME, false, false, false, null);

         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_NAME);

         channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, ROUTING_KEY_NAME);

         for (int i = 0; i < 10; i++) {

             String message = "死信交换机" + i;

             channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,

                     message.getBytes());

         }

         RabbitMqUtils.close(connection, channel);

     }

     private static Map<String, Object> getArguments() {

         Map<String, Object> result = new HashMap<String, Object>(5);

         result.put("x-message-ttl", 5000);

         result.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);

         result.put("x-dead-letter-routing-key", ROUTING_KEY_NAME);

         return result;

     }

 }

x-dead-letter-exchange参数其实只是定义了一个交换机的名称，实际上你还需要定义一套mq的流程，如代码exchangeDeclare、queueDeclare、queueBind。

x-dead-letter-routing-key参数的routing-key需要与之前的路由key一致，否则数据将会匹配不到，也就不会走到DEAD_EXCHANGE_NAME里去了。

发送消息：

void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body) throws IOException;

发送消息主要参数：mandatory、immediate。

1、mandatory：当消息无法路由到队列的处理方式。

2、immediate：当路由的队列无消费者时如何处理。

消费消息：

消费消息分为两种，一种是推，一种是主动拉。

1、推：basicConsumer（推，相当于监听，会消费调当前队列中所有的消息）

String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, Consumer callback) throws IOException;

autoAck：是否自动确认，true-自动确认，false-非自动确认；当autoAck=true时，消费完时就会自动删除这条消息。

noLocal：设置为ture时，不能将同一个connection生产的消息在此connection消费，也就是说一个connection不能同时为生产者和消费者。

2、拉：basicGet（拉，只会拉取一条消息）

GetResponse basicGet(String queue, boolean autoAck) throws IOException;

注意：不能用for循环的basicGet代替basicConsumer，因为这样性能非常差。

确认与拒绝：

1、消息确认：

）概念解释：

为了保证消息可靠的到达消费者，RabbitMQ提供了消息确认机制。

首先我们要知道除了手动的确认消息，还可以通过basicConsumer指定的autoAck参数来确认消息的到达。

当autoAck=false时，RabbitMQ会显式的等待消息回复确认后，才会将消息从内存或磁盘中移除（先设置为移除标记，然后在真正的移除）；当autoAck=true时，RabbitMQ不管消费者有没有接收到消息，都会直接将消息移除。

注意：当RabbitMQ一直没有收到消费者确认消息的通知，并且消费者的连接断开时，那么此条消息会重新进入队列。

void basicAck(long deliveryTag, boolean multiple) throws IOException;

multiple：是否确认多条消息；true-确认多条（确认该信道上deliveryTag标记之前所有未经确认的消息），false-确认单条。

）代码示例：

 public class AckProduct {

     private static final String EXCHANGE_NAME = "ack.exchange";

     private static final String QUEUE_NAME = "ack.queue";

     private static final String ROUTING_KEY = "ack.routing-key";

     public static void main(String[] args) throws IOException, TimeoutException {

         Connection connection = RabbitMqUtils.getConnection();

         Channel channel = connection.createChannel();

         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

         for (int i = 0; i < 20; i++) {

             channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, ("ack" + i).getBytes());

         }

         RabbitMqUtils.close(connection, channel);

     }

 }

 public class AckConsumer {

     private static final String QUEUE_NAME = "ack.queue";

     public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

         Connection connection = RabbitMqUtils.getConnection();

         final Channel channel = connection.createChannel();

         // 因为basicConsumer会消费队列中所有的消息，这样不方便演示basicAck中multiple参数的效果，故用basicGet来消费消息

         GetResponse getResponse = channel.basicGet(QUEUE_NAME, false);

         System.err.println(getResponse.getEnvelope().getDeliveryTag() + ": " + new String(getResponse.getBody()));

         System.err.println("-------------------------");

         GetResponse getResponse2 = channel.basicGet(QUEUE_NAME, false);

         System.err.println(getResponse2.getEnvelope().getDeliveryTag() + ": " + new String(getResponse2.getBody()));

         System.err.println("-------------------------");

         TimeUnit.SECONDS.sleep(10);

         GetResponse getResponse3 = channel.basicGet(QUEUE_NAME, false);

         System.err.println(getResponse3.getEnvelope().getDeliveryTag() + ": " + new String(getResponse3.getBody()));

         channel.basicAck(getResponse3.getEnvelope().getDeliveryTag(), false);

         System.err.println("-------------------------");

         // 休眠10秒，方便看到multiple参数的效果

         TimeUnit.SECONDS.sleep(10);

         GetResponse getResponse4 = channel.basicGet(QUEUE_NAME, false);

         System.err.println(getResponse4.getEnvelope().getDeliveryTag() + ": " + new String(getResponse4.getBody()));

         channel.basicAck(getResponse4.getEnvelope().getDeliveryTag(), true);

         RabbitMqUtils.close(connection, channel);

     }

 }

首先AckProduct生产20条消息，AckConsumer先消费两条消息，再休眠10秒，此时可以看到ready为18，unacked为2，这两消息未确认（10 - 18行）；

然后我们确认一条消息后，发现ready为17，unacked为2（20 - 23行）；

此时便是关键时刻，若multiple真是如其定义一般，那么在执行28 - 30行后，ready应该是16，而未确认消息unacked应该是0才对，10秒后发现果然如此。

2、消息拒绝：

）拒绝单条消息：

void basicReject(long deliveryTag, boolean requeue) throws IOException;

）拒绝多条消息：

void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;

）参数说明：

multiple：是否拒绝多条消息；true-拒绝多条（拒绝该信道上deliveryTag标记之前所有未经确认的消息），false-拒绝单条（与basicAck类似）。

requeue：是否将拒绝的消息重新放入队列中；true-重新放入，false-丢弃。

巴特西

RabbitMQ学习笔记（三、生产者与消费者）

目录：

细说交换器：

细说队列：

发送消息：

消费消息：

确认与拒绝：

最新文章

热门文章