RabbitMQ高级特性-阿里云开发者社区

Confirm确认消息

Confirm消息确认机制：了消息的确认，是指生产者投递消息后，如果 Broker 收到消息，则会给我们生产者一个应答。生产者进行接收应答，用来确定这条消息是否正常的发送到 Broker ，这种方式也是消息的可靠性投递的核心保障！

1.在 channel 上开启确认模式： channel . confirmselect ( )

2.在channel上添加监听： addConfirmListener ，监听成功和失败的返回结果，根据具体的结果对消息进行重新发送、或记录日志等后续处理！

生成者：

public class Producer {
  public static void main(String[] args) throws Exception {
    //1 创建ConnectionFactory
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.11.76");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    //2 获取Connection
    Connection connection = connectionFactory.newConnection();
    //3 通过Connection创建一个新的Channel
    Channel channel = connection.createChannel();
    //4 指定我们的消息投递模式: 消息的确认模式 
    channel.confirmSelect();
    String exchangeName = "test_confirm_exchange";
    String routingKey = "confirm.save";
    //5 发送一条消息
    String msg = "Hello RabbitMQ Send confirm message!";
    channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
    //6 添加一个确认监听
    channel.addConfirmListener(new ConfirmListener() {
      @Override
      public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        System.err.println("-------no ack!-----------");
      }
      @Override
      public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        System.err.println("-------ack!-----------");
      }
    });
  }
}

Return 消息机制

Return LiStener 用于处理一些不可路由的消息！

我们的消息生产者，通过指定一个 Exchange 和Routingkey ，把消息送达到某一个队列中去，然后我们的消费者监听队列，进行消费处理操作！

但是在某些情况下，如果我们在发送消息的时候，当前的Exchange不存在或者指定的路由 key 路由不到，这个时候如果我们需要监听这种不可达的消息，就要使用 Return Listener !

Return 消息机制在基础 Api 中有一个关键的配置项： Mandatory ：如果为 true ，则监听器会接收到路由不可达的消息然后进行后续处理，如果为 false ，那么 broker 端自动删除该消息！

消费者：

public class Consumer {
  public static void main(String[] args) throws Exception {
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.11.76");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    Connection connection = connectionFactory.newConnection();
    Channel channel = connection.createChannel();
    String exchangeName = "test_return_exchange";
    String routingKey = "return.#";
    String queueName = "test_return_queue";
    channel.exchangeDeclare(exchangeName, "topic", true, false, null);
    channel.queueDeclare(queueName, true, false, false, null);
    channel.queueBind(queueName, exchangeName, routingKey);
    QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
    channel.basicConsume(queueName, true, queueingConsumer);
    while(true){
      Delivery delivery = queueingConsumer.nextDelivery();
      String msg = new String(delivery.getBody());
      System.err.println("消费者: " + msg);
    }
  }
}

生产端：

public class Producer {
  public static void main(String[] args) throws Exception {
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.254.129");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    Connection connection = connectionFactory.newConnection();
    Channel channel = connection.createChannel();
    String exchange = "test_return_exchange";
    String routingKey = "return.save";
    String routingKeyError = "abc.save";
    String msg = "Hello RabbitMQ Return Message";
    channel.addReturnListener(new ReturnListener() {
      @Override
      public void handleReturn(int replyCode, String replyText, String exchange,
          String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("---------handle  return----------");
        System.err.println("replyCode: " + replyCode);
        System.err.println("replyText: " + replyText);
        System.err.println("exchange: " + exchange);
        System.err.println("routingKey: " + routingKey);
        System.err.println("properties: " + properties);
        System.err.println("body: " + new String(body));
      }
    });
        //第三个参数为mandatory
    channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
    //channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
  }
}

自定义消费者使用

继承DefaultConsumer重写handleDelivery方法。

public class MyConsumer extends DefaultConsumer {
  public MyConsumer(Channel channel) {
    super(channel);
  }
  @Override
  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.err.println("-----------consume message----------");
    System.err.println("consumerTag: " + consumerTag);
    System.err.println("envelope: " + envelope);
    System.err.println("properties: " + properties);
    System.err.println("body: " + new String(body));
  }
}

channel.basicConsume(queueName, true, new MyConsumer(channel));

消费端限流

当我们的RabbitMQ上有成千上万未处理的消息，单个客户端没有办法处理这么多消息。

所有我们一般都不会去使用自动签收（autoAck:true）

RabbitMQ提供了一种qos (服务质量保证)功能，即在非自动确认消息的前提下，如果一定数目的消息(通过基于consume或者channel设置Qos的值)未被确认前，不进行消费新的消息。

void basicQos(int prefetchSize, int prefetchCount, boolean global)

prefetchSize: 0 消息的大小限制设置为0表示不做限制。

prefetchCount:会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack,则该consumer将block掉，直到有消息ack。（每次消费几条数据，消费完就发送ack，在传过来几条）

global: true\false 是否将上面设置应用于channel

简单点说，就是上面限制是channel级别的还是consumer级别

每次签收一条

    //1 限流方式  第一件事就是 autoAck设置为 false
    channel.basicQos(0, 1, false);

自定义消费者ack

public class MyConsumer extends DefaultConsumer {
  private Channel channel ;
  public MyConsumer(Channel channel) {
    super(channel);
    this.channel = channel;
  }
  @Override
  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.err.println("-----------consume message----------");
    System.err.println("consumerTag: " + consumerTag);
    System.err.println("envelope: " + envelope);
    System.err.println("properties: " + properties);
    System.err.println("body: " + new String(body));
        //1.消息标签  2.是否批量签收
    channel.basicAck(envelope.getDeliveryTag(), false);
  }
}

Dead-Letters-Queue “死信队列”

来自一个队列的消息可以被当做‘死信’，即被重新发布到另外一个“exchange”去，这样的情况有：

消息被拒绝 (basic.reject or basic.nack) 且带 requeue=false 参数
消息的TTL-存活时间已经过期
队列长度限制被超越（队列满）

Dead Letter Pattern

“死信”模式指的是，当消费者不能处理接收到的消息时，将这个消息重新发布到另外一个队列中，等待重试或者人工干预。

这个过程中的exchange和queue就是所谓的"Dead Letter Exchange 和 Queue"。

关键是如何区分“消费失败”和“处理失败”？消费失败需要送到死信队列，而处理失败不需要。大部分情况都属于“处理失败”。

生产端

public class Producer {
  public static void main(String[] args) throws Exception {
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.254.129");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    Connection connection = connectionFactory.newConnection();
    Channel channel = connection.createChannel();
    String exchange = "test_dlx_exchange";
    String routingKey = "dlx.save";
    String msg = "Hello RabbitMQ DLX Message";
    for(int i =0; i<1; i ++){
      //设置过期时间
      AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
          .deliveryMode(2)
          .contentEncoding("UTF-8")
          .expiration("10000")
          .build();
      channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
    }
  }
}

消费端

public class Consumer {
  public static void main(String[] args) throws Exception {
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.254.129");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    Connection connection = connectionFactory.newConnection();
    Channel channel = connection.createChannel();
    // 这就是一个普通的交换机 和 队列 以及路由
    String exchangeName = "test_dlx_exchange";
    String routingKey = "dlx.#";
    String queueName = "test_dlx_queue";
    channel.exchangeDeclare(exchangeName, "topic", true, false, null);
    Map<String, Object> agruments = new HashMap<String, Object>();
    agruments.put("x-dead-letter-exchange", "dlx.exchange");
    //这个agruments属性，要设置到声明队列上
    channel.queueDeclare(queueName, true, false, false, agruments);
    channel.queueBind(queueName, exchangeName, routingKey);
    //要进行死信队列的声明:
    channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);
    channel.queueDeclare("dlx.queue", true, false, false, null);
    channel.queueBind("dlx.queue", "dlx.exchange", "#");
    channel.basicConsume(queueName, true, new MyConsumer(channel));
  }
}