首页 > 代码库 > RabbitMQ （消息队列）专题学习03 Work Queues(工作队列)

RabbitMQ （消息队列）专题学习03 Work Queues(工作队列)

2024-07-03 03:53:25   226人阅读

一、概述

工作队列(Work queues)

(使用Java客户端)

在前面的专题学习中，我们使用Java语言实现了一个简单的从名为"hello"的队列中发送和接收消息的程序，在这部内容中我们将创建一个工作队列被用来分配定时消息任务，而且通过多个接收者(工作者)实现。

工作队列(又名任务队列)，主要的思想是为了避免立即做一个资源密集型的任务(多消息同时密集发送)，不必等待它完成，当运行许多工作者的让任务都在她们之间共享。

它在web应用中是非常有用的，因为在很短的时间内http请求窗口处理一个复杂的任务是不可能实现的，它的结构如下图-1所示：

图-1

二、实现步骤

2.1、准备

在本部分内容之前，已经实现了发送单条”Hello World!“的消息，现在将发送一些复杂的字符串任务，由于没有一个真实的生产环境来模拟，我们可以通过使用Thread.sleep()函数来假设任务通过描述字符串hello...将要花费三秒钟的时间。

从之前的实例中，我们只需要稍微修改Sender01.java代码，允许任意的消息通过命令行发送，这一计划将安排到工作队列中的任务，让我们将

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();
}
之前的旧的Recv01.java也需要做一些改变。

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");
}
模拟假任务执行时间：

private static void doWork(String task) throws InterruptedException {
    for (char ch: task.toCharArray()) {
        if (ch == ‘.‘) Thread.sleep(1000);
    }
}

2.2、轮询调度

使用消息队列的优点之一就是能够方便地并行工作，如果我们建立了大量的工作任务，我们就可以添加更多的worker，这样大规模应用就显的比较容易，首先让我们尝试同时运行两个Worker.java脚本，他们都将获得队列中的消息，但是具体情况如下，我们需要打开三个控制台，两个将运行Worker.java的脚本，这些控制台将显示两个消费者C1和C2，第三个控制台将发布一个新的任务，一旦启动了消费者，你将可以通过第三个控制台发布一些消息，在默认情况下RabbitMQ将发送每条消息给下一个消费者，在队列里每个消费者将获得同样数量的消息，这种方式被称之轮询调度。

2.3、消息确认

执行一个任务要花费几秒钟，你可能想知道在此过程中会发生什么，如果一个消费者启动一项长期任务，它只是部分完成了，在我们当前的代码中，一旦RabbitMQ提供了一个消息给客户立即就将其从内存中删除，在这种情况下，如果一个工作者down掉，我们就将失去消息，尽管它只是处理，但是也将失去这部分消息，但是我们必须确保不丢失任何消息，如果一个工作者down了，我们希望这部分任务被传递到另外一个工作者的任务当中。

为了确保每条消息不丢失，RabbitMQ支持消息的确认(Acknowledgments)，一个确认被送回消费者告诉RabbitMQ的一个特定的消息一经被接收和处理，RabbitMQ此时才可以将该特定的消息删除。

如果消费者进程被杀掉而没有发送一个确认给RabbitMQ服务器，RabbitMQ会明白这个消息是没有被正常完成处理。

目前还有任何的消息超市，只有当工作者连接中断,RabbitMQ将重新传送消息，即使重新处理消息需要一个很长的时间。

消息确认默认是被打开的，在之前的实例中，我们可以通过AUTOACK=true来打开这个标记，当我们完成了任务，就应该移除这个标志，并且发送一个确认给工作者(Worker).

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

2.4、消息持久化

前面已经给出了如何确保消费者down了，任务也不会丢失，但是我们的任务在RabbitMQ服务器停止时还是可能会被丢失，当RabbitMQ服务器宕机或者奔溃时，它会丢失所有的队列和消息，除非告诉它不要这么做，需要做两件事情确保消息不会被丢失，我们需要标记队列和消息持久化，首先我们需要确保RabbitMQ永远不会丢失队列，为了做到这点，我们需要将队列声明为持久化：

boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);
此时queueDeclare变化需要被兼容到生产者和消费者的代码中，在这一点上，我们肯定不会丢失队列，即使task_queue队列所属的RabbitMQ服务器重新启动，现在我们需要将消息标记为持久性，通过设置MessageProperties值为PERSISTENT_TEXT_PLAN来实现。

import com.rabbitmq.client.MessageProperties;
channel.basicPublish("", "task_queue", 
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());

消息持久化需要注意的事项：

标记为持久化的消息并不能完全保证消息不会被丢失，虽然它告诉RabbitMQ的消息保存到磁盘，仍然有一个很短的时间内，RabbitMQ的消息一经接收了，但是并没有成功保存到磁盘，而是保存在缓存汇总，持久化的保证能力不足，但它是我们简单任务队列已经足以满足需求了。

2.5、公平调度

RabbitMQ当消息进入队列后，仅仅只是分发消息，它没有为消费者查找未确定的消息，它只是一味的分发n个消息给n个消费者，假如我需要两个工作者来分别处理消息，一个工作者处理所有奇数的消息，另外一个工作者处理偶数的消息，此时就需要打破RabbitMQ广播式的分发消息的规则，我们可以使用basicQos方法来实现，将其设置为1。

int prefetchCount = 1;
channel.basicQos(prefetchCount);
它的结构图如下图-2所示：

图-2

2.6、完整的代码清单：

package com.xuz.task;
import java.io.IOException;
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";
	public static void main(String[] args) throws IOException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("127.0.0.1");
		Connection conn = factory.newConnection();
		Channel channel = conn.createChannel();
		/**
		 * true：消息持久化设置
		 * 这样设置之后，服务器收到消息后就会立刻将消息写入到硬盘，就可以防止突然服务器挂掉，而引起的数据丢失了。
		 * 但是服务器如果刚收到消息，还没来得及写入到硬盘，就挂掉了，这样还是无法避免消息的丢失。
		 */
		channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
		//封装发送消息
//		String[] msg={"xuzheng","test01","rabbitMQ"};
//		String message = getMessage(msg);
		String message = null;
		for (int i = 0; i < 10; i++) {
			message = "测试公平调度"+i;
			//消息持久化设置MessageProperties.PERSISTENT_TEXT_PLAIN 当rabbitMQ暂时down掉，下次重启之后，工作者还是能接受目前发送的消息。
			channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,
					message.getBytes());
			System.out.println("send:["+message+"]");
		}
		channel.close();
		conn.close();
	}

	private static String getMessage(String[] args) {
		if(args.length<1){
			return "Hello World!";
		}else{
			return joinStrings(args,"");
		}
	}

	private static String joinStrings(String[] args, String string) {
		int len = args.length;
		if(len == 0){
			return "";
		}
		StringBuilder words = new StringBuilder(args[0]);
		for (int i = 0; i < len; i++) {
			words.append(string).append(args[i]);
		}
		return words.toString();
	}
}	


package com.xuz.work;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;
public class Worker {
	private static final String TASK_QUEUE_NAME = "task_queue";
	public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("127.0.0.1");
		Connection conn = factory.newConnection();
		Channel channel = conn.createChannel();
		/**
		 * true:
		 * 这样设置之后，服务器收到消息后就会立刻将消息写入到硬盘，就可以防止突然服务器挂掉，而引起数据丢失了，但是服务器如果刚收到消息，还没有来得写入硬盘，就挂掉了，这样
		 * 无法避免消息得丢失。
		 */
		channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
		System.out.println("waiting for message.To exit press CTRL+C");
		channel.basicQos(1);
		QueueingConsumer consumer = new QueueingConsumer(channel);
		/**
		 * false：设置确认消息，true表示接收到消息之后，将返回给服务端确定消息
		 */
		channel.basicConsume(TASK_QUEUE_NAME, false,consumer);
		while(true){
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			String message = new String(delivery.getBody());
			System.out.println("Received:["+message+"] from Task");
			doWork(message);
			System.out.println("Done!");
			//设置消息确认机制，如将如下代码注释掉，则
			//一旦将autoAck关闭之后，一定要记得处理完消息之后，向服务器确认消息。否则服务器将会一直转发该消息
			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
		}
	}
	private static void doWork(String message) throws InterruptedException {
		for (char ch : message.toCharArray()) {
			if(ch == ‘.‘)Thread.sleep(1000);
		} 
	}
}


2.7、测试消息确认

为了确保消息一定被消费者处理，RabbitMQ提供了消息确认功能，就是在消费者处理完任务之后，就给服务器一个确认消息，服务器就会将该消息删除，如果消费者超时不反馈确认消息，那么服务器就将该消息重新发送给其他消费者，默认是开启的。

操作步骤：

1、运行NewTask.java,发送消息

2、运行Worker.java接收消息

3、再次运行Worker.java来接收消息，此时接收不到消息了(因为第一次接收成功之后，客户端会给服务器反馈确认消息，服务器接收到确认之后将消息从队列中删除掉)

此时没有接收到服务端的发送的消息，因为通过如下代码实现了接收者在每次接收到消息之后向服务器发送确认消息之后，服务器就不会再发送同样的消息给接收者,若将如下代码注释之后，继续测试

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
重复上述三步，观察结果如下：

 

此时因为在客户端接收消息之后没有给服务器反馈确认信息，服务器没有收到确认信息就不会从队列中删除消息，此时客户端仍然能从队列中获取消息

2.8、持久化测试

如果RabbitMQ-server突然宕机，那么还没有被读取的消息还是会丢失，所以我们可以让消息持久化，只需要在定义queue的时候，设置持久化消息即可，方法如下：

channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
这样设置后，服务器收到消息后就会立刻将消息写入到磁盘，就可以防止突然宕机引起数据丢失，但是服务器如果刚收到消息时，还没来的急写入磁盘就宕机了，这样还是无法避免消息的丢失。

操作步骤(代码使用公平调度修改后的代码)：

1、运行NewTask.java

2、停止RabbitMQ-server服务：开始---程序---RabbitMQ Server--RabbitMQ Service - stop

3、运行work.java(因为RabbitMQ已经停止了，此时提示拒绝连接)

4、启动RabbitMQ-Server服务：开始---程序---RabbitMQ Server--RabbitMQ Service - start

5、运行work.java

此时仍然能接收到消息，说明通过持久化设置后，服务器将要发送的消息写入到了磁盘中，并不是缓存当中，当下次服务器启动之后，不运行发送者，接收者也可以从指定的队列中取出从存储在磁盘中的消息。

2.9、公平调度测试

从worker看出，必须处理完一个消息，才会取接收下一个消息，如果生产者众多，那么一个消费者肯定忙不过来，此时就可以用多个消费者来对同一个channel的消息进行处理，并且要公平分配任务给多个消费者，不能部分消费者很忙，部分消费者很闲，实现公平调度的方式在前面已经提到了，设置basicQos为1即可。

操作步骤：

1、运行Worker.java如下：

2、运行Worker01.java如下：

3、运行NewTask.java，会看到worker.java的控制台和worker01.java的控制台处会有消息打印出来，如下：

 http://

从上面的结果中可以看出RabbitMQ给每个工作者分别处理了5个不同的工作任务，体现其公平调度的原则。

至此基于RabbitMQ实现的工作队列相关实例全部完成！

源码下载

基于RabbitMQ 工作队列实现


联系
我们