rabbitMq 直连模式实现多消费者轮询消费（一对多）

今天通过rabbitMq的直连模式（direct）来实现多个消费者对消息的轮询读取

前提是已经安装了rabbitMq！

话不多说，步入正题：

首先创建一个连接mq的服务器：

ConnectionUtil ：

public class ConnectionUtil {

    public static Connection getConnection()throws Exception{
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();

        //配置参数
        connectionFactory.setHost("localhost");//主机IP地址
        connectionFactory.setPort(5672);//端口
        connectionFactory.setVirtualHost("/");//虚拟机，默认/
        connectionFactory.setUsername("test");//用户名
        connectionFactory.setPassword("test");//密码

        //通过工厂创建连接
        Connection connection = connectionFactory.newConnection();
        return connection;
    }
}

生产者：Producer

public class Producer {

    static final String QUEUE_NAME = "work_queue";//队列名


    public static void main(String[] args) throws Exception {

        //获取连接
        Connection connection = ConnectionUtil.getConnection();

        //创建频道
        Channel channel = connection.createChannel();
        channel.queueBind(QUEUE_NAME, "amq.direct", "message_ttl_routingKey");


        for (int i = 0; i < 10; i++) {
            String message = i + "队列消息~~~";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        }

        //关闭频道
        channel.close();
        //关闭连接
        connection.close();
    }
}

消费者：Consumer

public class Consumer {

    static final String QUEUE_NAME="work_queue";


    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();

        //使用连接创建频道
        Channel channel = connection.createChannel();

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

        //接收消息(消费端监听)
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){

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

                long deliveryTag = envelope.getDeliveryTag();
                System.out.println("deliverTag: "+deliveryTag);
                System.out.println("consumerTag："+consumerTag);

                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);
                System.out.println("body："+new String(body));
            }
        };

        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);
    }
}

消费者2：Consumer2

public class Consumer2 {
    static final String QUEUE_NAME="work_queue";


    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();

        //使用连接创建频道
        Channel channel = connection.createChannel();

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

        //接收消息(消费端监听)
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){

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

                long deliveryTag = envelope.getDeliveryTag();
                System.out.println("deliverTag: "+deliveryTag);

                System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);
                System.out.println("body："+new String(body));
            }
        };

        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);
    }
}

然后启动消费者，再启动生产者：结果如下：

消费者1：

消费者2：