Work Queues(点对多)

Work Queues(点对多)

多个消费者在同一个消息队列中获取消息的情况。在有些应用当中，消费端接收到消息任务需要长时间的处理，如果等上一个消息处理完成以后再取下一个数据进行处理的话，势必会有一些延迟。在消息队列中的数据也会不断增多，延迟将越来越大。当然对于一个消费进程来说，在某些情况下可以起多个线程来处理，而在这里将介绍另一种处理方式，多个消费进程的情况。而RabbitMQ在这方面进行了很好的处理和封装，使客户程序可以很方便的使用。

默认的，RabbitMQ会顺序的把消息发送到下一个Consumer，这种发送消息的方式称为循环发送(round-robin)

package com.rabbitmq.www.work_queues;



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

public class NewTask {

  private static final String TASK_QUEUE_NAME = "task_queue";
  private final static String HOST_ADDR = "172.18.112.102";

  public static void main(String[] argv) throws Exception {
      
      
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost(HOST_ADDR);
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();
    
    //durable 设置true,queue持久化，server重启，此queue不丢失
    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
    for(int i=0;i<=10;i++){
        String message = "hello world";
        message=message+i;
        //BasicProperties设置MessageProperties.PERSISTENT_TEXT_PLAIN，信息持久化
        channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message);
    }

    channel.close();
    connection.close();
  }

}

package com.rabbitmq.www.work_queues;

import com.rabbitmq.client.*;

import java.io.IOException;

public class Worker {

  private static final String TASK_QUEUE_NAME = "task_queue";
  private final static String HOST_ADDR = "172.18.112.102";

  public static void main(String[] argv) throws Exception {
      
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost(HOST_ADDR);
    final Connection connection = factory.newConnection();
    final Channel channel = connection.createChannel();
  //durable 设置true,queue持久化，server重启，此queue不丢失
    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    //一次只接受一条信息
    channel.basicQos(1);

    final Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");

        System.out.println(" [x] Received '" + message + "'");
        try {
          doWork(message);
        } finally {
          System.out.println(" [x] Done");
          //向服务器发送应答
          channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };
    //autoAck 设置false,消费端挂掉，信息不会丢失，server会re-queue
    channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
  }

  private static void doWork(String task) {
    for (char ch : task.toCharArray()) {
      if (ch == '.') {
        try {
          Thread.sleep(1000);
        } catch (InterruptedException _ignored) {
          Thread.currentThread().interrupt();
        }
      }
    }
  }
}