Rabbitmq之高级特性——实现消费端限流&NACK重回队列_dhtqr-程序员宅基地

　　如果是高并发下，rabbitmq服务器上收到成千上万条消息，那么当打开消费端时，这些消息必定喷涌而来，导致消费端消费不过来甚至挂掉都有可能。

在非自动确认的模式下，可以采用限流模式，rabbitmq 提供了服务质量保障qos机制来控制一次消费消息数量。

下面直接上代码：

生产端：

package com.zxy.demo.rabbitmq;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ReturnListener;
import com.rabbitmq.client.AMQP.BasicProperties;

public class Producter {

    public static void main(String[] args) throws IOException, TimeoutException {
        // TODO Auto-generated method stub
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.10.110");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        Connection conn = factory.newConnection();
        Channel channel = conn.createChannel();
        String exchange001 = "exchange_001";
        String queue001 = "queue_001";
        String routingkey = "mq.topic";
        String body = "hello rabbitmq!===============限流策略";
//        开启确认模式
        channel.confirmSelect();
//        循环发送多条消息        
        for(int i = 0 ;i<10;i++){
        channel.basicPublish(exchange001, routingkey, null, body.getBytes());
    }
        
//        添加一个返回监听========消息返回模式重要添加
        channel.addConfirmListener(new ConfirmListener() {
            
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("===========NACK============");
                
            }
            
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("===========ACK============");
                
            }
        });
    }

}

 

消费端：

package com.zxy.demo.rabbitmq;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Receiver {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // TODO Auto-generated method stub
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.10.110");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        Connection conn = factory.newConnection();
        Channel channel = conn.createChannel();
        String exchange001 = "exchange_001";
        String queue001 = "queue_001";
        String routingkey = "mq.*";
        channel.exchangeDeclare(exchange001, "topic", true, false, null);
        channel.queueDeclare(queue001, true, false, false, null);
        channel.queueBind(queue001, exchange001, routingkey);
//        设置限流策略
//        channel.basicQos(获取消息最大数[0-无限制], 依次获取数量, 作用域[true作用于整个channel，false作用于具体消费者]);
        channel.basicQos(0, 2, false);
//        自定义消费者
        MyConsumer myConsumer = new MyConsumer(channel);
//        进行消费，签收模式一定要为手动签收
        Thread.sleep(3000);
        channel.basicConsume(queue001, false, myConsumer);
    }

}

 

自定义消费者：

package com.zxy.demo.rabbitmq;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 可以继承，可以实现，实现的话要覆写的方法比较多，所以这里用了继承
 *
 */
public class MyConsumer extends DefaultConsumer{
    private Channel channel;
    public MyConsumer(Channel channel) {
        super(channel);
        // TODO Auto-generated constructor stub
        this.channel=channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
            throws IOException {
        System.out.println("消费标签："+consumerTag);
        System.out.println("envelope.getDeliveryTag():==="+envelope.getDeliveryTag());
        System.out.println("envelope.getExchange():==="+envelope.getExchange());
        System.out.println("envelope.getRoutingKey():==="+envelope.getRoutingKey());
        System.out.println("body:==="+new String(body));
//        手动签收，一定要有消费者签收,如果没有如下代码，则限流模式下，仅能打印出来channel.basicQos(0, 2, false);第二参数的2条信息
        channel.basicAck(envelope.getDeliveryTag(), false);
    }
    

}

 重回队列模式，是当投递消息失败时，让该消息重新回到队列的模式，该模式需要手动签收，并需要在消费者中进行判断，调用重回队列的确认模式

代码如下

生产端：

package com.zxy.demo.rabbitmq;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import org.springframework.amqp.core.Message;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ReturnListener;

public class Producter {

    public static void main(String[] args) throws IOException, TimeoutException {
        // TODO Auto-generated method stub
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.10.110");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        Connection conn = factory.newConnection();
        Channel channel = conn.createChannel();
        String exchange001 = "exchange_001";
        String queue001 = "queue_001";
        String routingkey = "mq.topic";
        
//        循环发送多条消息        
        for(int i = 0 ;i<5;i++){
            String body = "hello rabbitmq!===============ACK&重回队列,第"+i+"条";
            Map<String,Object> head = new HashMap<>();
            head.put("n", i);
            BasicProperties properties = new BasicProperties(null, "utf-8", head, 2, 1, null, null, null, null, null, null, null, null, null);
            
        channel.basicPublish(exchange001, routingkey, properties, body.getBytes());
    }
        
    }

}

 

消费端：

package com.zxy.demo.rabbitmq;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Receiver {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // TODO Auto-generated method stub
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.10.110");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        Connection conn = factory.newConnection();
        Channel channel = conn.createChannel();
        String exchange001 = "exchange_001";
        String queue001 = "queue_001";
        String routingkey = "mq.*";
        channel.exchangeDeclare(exchange001, "topic", true, false, null);
        channel.queueDeclare(queue001, true, false, false, null);
        channel.queueBind(queue001, exchange001, routingkey);
//        自定义消费者
        MyConsumer myConsumer = new MyConsumer(channel);
//        进行消费，签收模式一定要为手动签收
        channel.basicConsume(queue001, false, myConsumer);
    }

}

 

自定义消费者：

package com.zxy.demo.rabbitmq;

import java.io.IOException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 可以继承，可以实现，实现的话要覆写的方法比较多，所以这里用了继承
 *
 */
public class MyConsumer extends DefaultConsumer{
    private Channel channel;
    public MyConsumer(Channel channel) {
        super(channel);
        // TODO Auto-generated constructor stub
        this.channel=channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
            throws IOException {
        System.out.println("消费标签："+consumerTag);
        System.out.println("envelope.getDeliveryTag():==="+envelope.getDeliveryTag());
        System.out.println("envelope.getExchange():==="+envelope.getExchange());
        System.out.println("envelope.getRoutingKey():==="+envelope.getRoutingKey());
        System.out.println("body:==="+new String(body));
        System.out.println("===================休眠以便查看===============");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
//        手动签收
        Integer i = (Integer) properties.getHeaders().get("n");
        System.out.println("iiiiiiiiiiiiiiiii======================================================"+i);
        if(i==1) {
            channel.basicNack(envelope.getDeliveryTag(),false, true);//第三个参数为是否重返队列
        }else {
            channel.basicAck(envelope.getDeliveryTag(), false);    
        }
    }
    

}

下面是重回队列执行结果，可以看到当消费完后第一条不断的被扔回队列然后消费再扔回。

消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===1
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第0条
===================休眠以便查看===============
iiiiiiiiiiiiiiiii======================================================0
消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===2
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第1条
===================休眠以便查看===============
iiiiiiiiiiiiiiiii======================================================1
消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===3
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第2条
===================休眠以便查看===============
iiiiiiiiiiiiiiiii======================================================2
消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===4
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第3条
===================休眠以便查看===============
iiiiiiiiiiiiiiiii======================================================3
消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===5
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第4条
===================休眠以便查看===============
iiiiiiiiiiiiiiiii======================================================4
消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===6
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第1条
===================休眠以便查看===============
iiiiiiiiiiiiiiiii======================================================1
消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===7
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第1条
===================休眠以便查看===============
iiiiiiiiiiiiiiiii======================================================1
消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===8
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第1条
===================休眠以便查看===============
iiiiiiiiiiiiiiiii======================================================1
消费标签：amq.ctag-eeeaa9jgrJ3tDhtQR2XReg
envelope.getDeliveryTag():===9
envelope.getExchange():===exchange_001
envelope.getRoutingKey():===mq.topic
body:===hello rabbitmq!===============ACK&重回队列,第1条
===================休眠以便查看===============