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();
}
}
}
}
}