RabbitMQ学习笔记2-Work queues

2023-05-16

接下来学习第二种模型，Work queues模型，如图所示：

该模型描述的是：一个生产者（P）向队列发送一个消息，然后多个消费者（P）接受消息，每条消息只能被一个消费者接收。以下示例可根据学习笔记1里面的代码稍作修改，因此部分细节不再说明。

生产者发送消息到队列中

创建连接和通道

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

声明一个队列

channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

注意：此时第二个参数设置为true。作用是，当RabbitMQ退出或崩溃时，它会丢失队列和消息，。需要两件事来确保消息不会丢失：我们需要将队列和消息设置为持久化。

发送消息到队列

String message = "Hello World......";
// PERSISTENT_TEXT_PLAIN: Content-type "text/plain", deliveryMode 2 (persistent), priority zero
channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));

注意：参数3MessageProperties.PERSISTENT_TEXT_PLAIN作用是使消息持久化。

关闭通道和连接

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

源代码如下：

import java.io.IOException;
import java.util.concurrent.TimeoutException;

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

/**
 * @author YKun
 * @date 2017年4月24日 下午10:37:20
 * @describe
 */
public class NewTask {

	private static final String TASK_QUEUE_NAME = "task_queue";

	public static void main(String[] args) throws IOException, TimeoutException {

		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("127.0.0.1");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		
		// param1: 队列名 param2: 是否持久化
		channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

		String message = "Hello World......";
		
		// PERSISTENT_TEXT_PLAIN: Content-type "text/plain", deliveryMode 2 (persistent), priority zero
		channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
		
		System.out.println(" [x] Sent '" + message + "'");

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

}

运行结果如下：

消费者获取消息

创建连接和通道

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

声明一个队列

channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

定义一个默认的消费者

Consumer consumer = new DefaultConsumer(channel){
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, 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);
		        }
			}
		};

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

函数doWork的作用是跟据获取的消息，若消息内容存在"."则程序休眠一秒，channel.basicAck的作用是在处理完消息之后返回应答状态，第二个参数false表示关闭RabbitMQ的自动应答，改为手动应答。

监听队列，并且设置成手动返回

channel.basicConsume(TASK_QUEUE_NAME, false, consumer);

参数2，autoAck设置为false，表示不自动删除

Message acknowledgment 消息应答

在现实场景中经常会发生一种情况，其中一个消费者在执行任务过程中有可能会挂掉或者出现异常，可是RabbitMQ默认一旦将任务发送给消费者之后就会将该任务从内存中删除，因此极大可能会出现任务丢失的情况，所以当其中一个消费者宕机之后应该把这个任务转发给下一个消费者进行处理。为确保消息永不丢失，RabbitMQ支持消息确认，既消费者处理完任务后发送一个确认信息给RabbitMQ，RabbitMQ收到确认信息后方可删除该任务。若消费者出现宕机而不发送确认信息，RabbitMQ则会将任务重新排队交由下一个消费者处理。因此才会出现以下代码

channel.basicConsume(TASK_QUEUE_NAME, false, consumer);

Fair dispatch 公平派遣

当消息进入RabbitMQ时，RabbitMQ只会分派消息，而且是固定的将第n个消息分发给第n个消费者。可能会出现一种情况，例如假设有两个消费者，三个任务，第一个任务需要耗费20秒的时间进行处理，第二个任务只需5秒，第三个任务也只需5秒，则会出现以下情况，第一第三个任务都会被分派给第一个消费者，而第二个消费者可能处理完第二个任务之后就空闲下来，但第一个消费者连第一个任务都还没处理完。最佳的情况是当两个消费者有任一一个消费者空闲下来就应该处理掉还没被消费的任务，既第二个消费者应该将第三个任务处理掉。我们可以使用 basicQos 方法与 prefetchCount = 1 设置。这告诉RabbitMQ不要一次给一个工作者多个消息。或者换句话说，在处理并确认前一个消息之前，不要向工作人员发送新消息。相反，它将发送到下一个还不忙的工作。

int prefetchCount = 1 ;
channel.basicQos（prefetchCount）;

源代码如下：

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;

/**
 * @author YKun
 * @date 2017年4月24日 下午10:56:34
 * @describe
 */
public class Worker {

	private static final String TASK_QUEUE_NAME = "task_queue";

	public static void main(String[] args) throws IOException, TimeoutException {

		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("127.0.0.1");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		System.out.println(channel.hashCode());
		channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

		channel.basicQos(1);
		
		Consumer consumer = new DefaultConsumer(channel){
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, 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);
		        }
			}
		};
		
		// param2: autoAck设置为false，表示不自动删除
		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();
				}
			}
		}
	}

}

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)