工作队列(使用Java客户端)
在web应用程序中,这个理念是特别有用的,你无法在一个短暂的http请求中处理一个复杂的任务。 准备
在先前的指南中,我们发送了一个包含"Hello World!“消息。现在我们将要发送一些字符串,用来代表复杂的任务。我们没有一个真实的任务,比如图片的调整大小或者pdf文件渲染,所以我们通过
我们将会轻量的修改我们以前例子中 String message = getMessage(argv); channel.basicPublish("", "hello", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); 一些帮助从命令行中获取消息参数: private static String getMessage(String[] strings){ if (strings.length < 1) return "Hello World!"; return joinStrings(strings, " "); } private static String joinStrings(String[] strings, String delimiter) { int length = strings.length; if (length == 0) return ""; StringBuilder words = new StringBuilder(strings[0]); for (int i = 1; i < length; i++) { words.append(delimiter).append(strings[i]); } return words.toString(); }
我们老的 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); doWork(message); System.out.println(" [x] Done"); } 我们伪装的任务中冒充执行时间:
在第一部分指南中那样编译它们(jar 文件需要再工作路径上): $ javac -cp rabbitmq-client.jar NewTask.java Worker.java 循环分派
使用任务队列的优势之一是我们是容易并行处理。如果我们正在处理一些堆积的文件的话,我们仅仅需要增加更多的工作者,通过这种方式我们是容易扩展的。 shell1$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar Worker [*] Waiting for messages. To exit press CTRL+C shell2$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar Worker [*] Waiting for messages. To exit press CTRL+C 在这第三个控制平台我们用来发布新的任务。一旦你启动消费者,你就可以发布消息了: shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar NewTask First message. shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar NewTask Second message.. shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar NewTask Third message... shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar NewTask Fourth message.... shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar NewTask Fifth message..... 让我们看看什么被投递到我们工作者那里: shell1$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar Worker [*] Waiting for messages. To exit press CTRL+C [x] Received 'First message.' [x] Received 'Third message...' [x] Received 'Fifth message.....' shell2$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar Worker [*] Waiting for messages. To exit press CTRL+C [x] Received 'Second message..' [x] Received 'Fourth message....' 默认情况想,RabbitMQ将会把每一个消息发送给下一个消费者。平均下来每个消费者获取的消息数量是相同的。这种分布式消息方式被称为轮询。试试三个或更多的工作者。 消息确认
处理一个任务可能花费数秒时间,你可能会好奇如果一个消费者开始一个长任务,并且在处理完成部分的情况下就死掉了会发生什么情况。就我们当前的代码 来说,一旦RabbitMQ将消息传递给消费者,它就会立即将消息从内存中删除。在这种情况下,如果你杀掉一个正在处理的工作者你会丢失它正在处理的消 息。我们也同时失去了已经分配给这个工作者并且没有开始处理的消息。 QueueingConsumer consumer = new QueueingConsumer(channel); boolean autoAck = false; channel.basicConsume("hello", autoAck, consumer); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); //... channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); }
使用这段代码,我们可以保证即使你将一个正在处理消息的工作者通过
消息持久化我们已经学习了如何在确定消费者是否已经死掉,并且保证任务不被丢失。但是如果RabbitMQ服务器停止,我们的任务依旧会丢失。 当RabbitMQ退出或者崩溃,它将会忘记这队列和消息,除非你告诉它不要这样做。两个事情需要做来保证消息不会丢失:我们标记队列和消息持久化。 首先,我们需要确保RabbitMQ不会丢失我们的队列,为了这样做,我们需要将它声明为持久化: boolean durable = true; channel.queueDeclare("hello", durable, false, false, null);
虽然这命令是正确的,但它不会立即在我们的程序里运行。那是因为我们已经定义了一个不持久化的 boolean durable = true; channel.queueDeclare("task_queue", durable, false, false, null);
这个 import com.rabbitmq.client.MessageProperties; channel.basicPublish("", "task_queue", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
公平分发
你可能注意到了,分发过程并没有如我们想的那样运作。例如,在一个情况下有两个工作者,当所有奇数消息是重的和所有偶数是轻的,一个工作者会一直忙碌下去,而另一个则会几乎不做什么事情。好吧,RabbitMQ不会在意那个事情,它会一直均匀的分发消息。
为了解决这个问题,我们可以使用 int prefetchCount = 1; channel.basicQos(prefetchCount);
把它们放在一起
我们的 import java.io.IOException; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Channel; import com.rabbitmq.client.MessageProperties; public class NewTask { private static final String TASK_QUEUE_NAME = "task_queue"; public static void main(String[] argv) throws java.io.IOException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); String message = getMessage(argv); channel.basicPublish( "", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } //... }
(NewTask.java source) java.lang.InterruptedException { import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Channel; import com.rabbitmq.client.QueueingConsumer; public class Worker { private static final String TASK_QUEUE_NAME = "task_queue"; public static void main(String[] argv) throws java.io.IOException, java.lang.InterruptedException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); channel.basicQos(1); QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(TASK_QUEUE_NAME, false, consumer); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); doWork(message); System.out.println(" [x] Done" ); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } //... }
(Worker.java source)
想要了解更多关于通道方法和消息属性,你可以浏览
现在我们可以移到指南3了,学习怎么样将相同的消息传递给多个消费者 转载请保留固定链接: https://linuxeye.com/Linux/RabbitMQ.html |