RabbitMQ 工作队列模式 Work Queue Demo

news2024/11/15 12:41:01

工作队列模式,一个消息只能有一个消费者消费

生产者发送20条消息

消费者有两个

第一个消费

睡一秒取一个

第二个睡2秒取

public class WorkConsumerTest1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1 创建连接工厂
        ConnectionFactory connectionFactory=new ConnectionFactory();
        //2 设置rabbitmq  ip地址
        connectionFactory.setHost("localhost");
        //3 创建连接对象   Conection对象
        Connection connection=connectionFactory.newConnection();
        //4 创建管道  Chanel
        Channel channel=connection.createChannel();
        //5 设置队列属性
        /**
         * 第一个参数:队列的名称
         * 第二个参数:队列是否要持久化
         * 第三个参数:是否排他性(是否在同一个Connection,如果设置为true,不同的Connection是获得不到消息的)
         * 第四个参数:是否自动删除消息
         * 第五个参数:是否要设置一些额外的参数
         */
        channel.queueDeclare("02-work",false,false,true,null);
        //channel.basicQos(1);
        //6 使用chanel 去 rabbitmq 获取消息进行消费
        /**
         * 第一个参数:队列的名称
         * 第二个参数:是否自动签收
         * 第三个参数:消息属性
         * 第四个参数:消息内容
         */
        channel.basicConsume("02-work", false,new DeliverCallback(){
            /**
             * 当消息从mq 中取出来了会回调这个方法
             * 消费者消费消息就在这个  handle中进行处理
             */
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消息中的内容为:"+new String(delivery.getBody()));
                //channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            }
        },new CancelCallback(){
            /**
             * 当消息取消了会回调这个方法
             */
            @Override
            public void handle(String s) throws IOException {
                System.out.println(111);
            }
        });
        //7 关闭消息   注意消费者 需要持续监听,不要关闭
        //channel.close();
        //connection.close();
    }
}