MQ下单队列多消费者实现

2022-09-28

高并发场景下，多线程的单消费者不一定满足业务需求，需要部署多消费者才能满足业务需求。

MQ下，多个消费者绑定同一个队列是无法同时消费的，一个消息只能被一个消费者消费。

因为 RabbitMQ 有队列的负载均衡机制，消息会被负载到每个消费者进行消费，无法同时被多个消费者消费，多个消费者监听同一个队列，RabbitMQ 会进行消息的轮询分发。

但轮询分发消息+自动确认机制（消费者拿到消息就删除队列中消息），在某消费者宕机的情况下是不可行的，会丢失消息。为实现这个容灾能力，需要实现手动确认消息+一个通道只放一个消息，即实现了消费者的能者多劳。

工具

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


public class ConnectionUtil {
    public static Connection getConnection(String host, int port, String vHost,
        String userName, String passWord) throws Exception {

        //1、定义连接工厂        
        ConnectionFactory factory = new ConnectionFactory();
        //2、设置服务器地址       
        factory.setHost(host);
        //3、设置端口     
        factory.setPort(port);
        //4、设置虚拟主机、用户名、密码  
        factory.setVirtualHost(vHost);
        factory.setUsername(userName);
        factory.setPassword(passWord);

        //5、通过连接工厂获取连接    
        Connection connection = factory.newConnection();

        return connection;
    }
}

生产者

import com.peabody.utils.ConnectionUtil;

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


public class Producer {
    private final static String EXCHANGE_NAME = "topic_exchange2";

    public static void main(String[] args) throws Exception {
        //1、获取连接       
        Connection connection = ConnectionUtil.getConnection("10.68.252.126",
                5672, "/", "guest", "guest");

        //2、声明信道      
        Channel channel = connection.createChannel();
        //3、声明交换器，类型为topic     
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        //4、创建消息    
        String message = "hello rabbitmq111";
        //5、发布消息        
        channel.basicPublish(EXCHANGE_NAME, "update.Name", null,
            message.getBytes());
        System.out.println("生产者发送" + message + "'");
        //6、关闭通道      
        channel.close();
        //7、关闭连接      
        connection.close();
    }
}

消费者

消费者1

import com.peabody.utils.ConnectionUtil;

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


public class Consumer1 {
    private final static String QUEUE_NAME = "topic_queue_4";
    private final static String EXCHANGE_NAME = "topic_exchange2";

    public static void main(String[] args) throws Exception {
        //1、获取连接       
        Connection connection = ConnectionUtil.getConnection("10.68.252.126",
                5672, "/", "guest", "guest");

        //2、声明通道       
        Channel channel = connection.createChannel();
        //3、声明队列     
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //4、绑定队列到交换机，指定路由key为update.#    
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update.#");
        //同一时刻服务器只会发送一条消息给消费者      
        channel.basicQos(1);

        //5、定义队列的消费者        
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        //6、监听队列,手动返回完成状态       
        channel.basicConsume(QUEUE_NAME, false, queueingConsumer);

        //6、获取消息     
        while (true) {
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 消费者1：" + message + "'");
            //消费者1接收一条消息后休眠10毫秒         
            Thread.sleep(10);
            //返回确认状态       
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

消费者2

import com.peabody.utils.ConnectionUtil;

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


public class Consumer2 {
    private final static String QUEUE_NAME = "topic_queue_4";
    private final static String EXCHANGE_NAME = "topic_exchange2";

    public static void main(String[] args) throws Exception {
        //1、获取连接       
        Connection connection = ConnectionUtil.getConnection("10.68.252.126",
                5672, "/", "guest", "guest");

        //2、声明通道      
        Channel channel = connection.createChannel();
        //3、声明队列      
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //4、绑定队列到交换机，指定路由key为select.#      
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update.#");
        //同一时刻服务器只会发送一条消息给消费者    
        channel.basicQos(1);

        //5、定义队列的消费者     
        //6、监听队列,手动返回完成状态     
        channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel){
          
          	//7、获取消息    
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              	String message = new String(body);
                System.out.println(" 消费者2：" + message + "'");
                //消费者2接收一条消息后休眠10毫秒        
                Thread.sleep(1000);
                //返回确认状态          
                channel.basicAck(envelope.getDeliveryTag(), false);
        });
    }
}