rabbitmq入门学习二

1. 竞争消费者模式(Work queues)

竞争消费者模型图：

                       

  work queues与基本消息模型相比，多了一个消费端，两个消费端共同消费同一个队列中的消息。

应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

1.1 编写工具类

将rabbitmq连接代码提取成工具类,减少代码冗余

[Java] 纯文本查看 复制代码

public class ConnectionUtil {[/align]    /**
     * 建立与RabbitMQ的连接
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception {
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("192.168.56.130");
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }
}

1.2 生产者

[Java] 纯文本查看 复制代码

public class WorkProducer {

    //队列名称
    private static final String QUEUE="work";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection=null;
        Channel channel=null;
        try {
            connection = ConnectionUtil.getConnection();
            channel  = connection.createChannel();
            channel.queueDeclare(QUEUE,true,false,false,null);
            //循环发送多条消息
            for (int i = 0; i < 50; i++) {
                // 消息内容
                String message = "task .. " + i;
                channel.basicPublish("", QUEUE, null, message.getBytes());
                System.out.println(" [x] send '" + message + "'");
                Thread.sleep(i * 2);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            channel.close();
            connection.close();
        }
    }
}

1.3 消费者1

[Java] 纯文本查看 复制代码

public class WorkConsumer1 {
    //队列名称
    private static final String QUEUE="work";
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE, true, false, false, null);
        //定义消费方法
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("receive message.." + msg);
            }
        };
        boolean autoAck = true;
        channel.basicConsume(QUEUE, autoAck, defaultConsumer);
    }
}

1.4 消费者2

消费者2和消费者1基本一样,唯一不同的是在获取消息后加了一个Thread.sleep(1000);模拟消费耗时

[Java] 纯文本查看 复制代码

public class WorkConsumer2 {
    //队列名称
    private static final String QUEUE="work";
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE, true, false, false, null);
        //定义消费方法
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("receive message.." + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        boolean autoAck = true;
        channel.basicConsume(QUEUE, autoAck, defaultConsumer);
    }
}

1.5测试和问题解决：

   1、分别启动消费者1和消费者2。

   2、启动生产者发送多个消息。

结果：

  1、一条消息只会被一个消费者接收；

  2、rabbit采用轮询的方式将消息是平均发送给消费者的；

  3、消费者在处理完某条消息后，才会收到下一条消息。

结果：

存在问题?

  1. 消费者2比消费者1的效率要低，一次任务的耗时较长，然而两人最终消费的消息数量是一样的

  2. 消费者1大量时间处于空闲状态，消费者2一直忙碌

  3. 这种情况它不是一个竞争关系，而是采用轮询的方式将消息是平均发送给消费者的。

问题解决：

  我们可以通过channel.basicQos(1);设置rabbitmq消费信息的条数来处理完成竞争消费模式，rabbitmq一次只会发送一条消息给该消费者,只有等消费者确认消费后才会发送第二条消息

修改代码如下：

[Java] 纯文本查看 复制代码

public class WorkConsumer2 {
    //队列名称
    private static final String QUEUE="work";
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE, true, false, false, null);

//设置每次消费消息条数
        channel.basicQos(1);
        //定义消费方法
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("【消费者2】receive: " + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        boolean autoAck = false;
        channel.basicConsume(QUEUE, autoAck, defaultConsumer);
    }
}

重新运行测试效果：

  从下图我们看以看出消费者2只消费了3条消息，消费者1消费了余下的所有消息，这才是竞争消费模式，处理快的多消费，处理慢的少消费。

看一看。