在RabbitMQ中实现Work queues工作队列模式

package com.itheima.rabbitmq.work; 
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.Channel; 
import com.rabbitmq.client.Connection; 
import com.rabbitmq.client.ConnectionFactory; 
public class Producer { 
	static final String QUEUE_NAME = "work_queue"; 
	public static void main(String[] args) throws Exception { 
		//创建连接 
		Connection connection = ConnectionUtil.getConnection(); 
		// 创建频道 
		Channel channel = connection.createChannel(); 
		// 声明（创建）队列 
		/**
		 * 参数1：队列名称 
		 * 参数2：是否定义持久化队列 
		 * 参数3：是否独占本次连接 
		 * 参数4：是否在不使用的时候自动删除队列 
		 * 参数5：队列其它参数 
		*/ 
		channel.queueDeclare(QUEUE_NAME, true, false, false, null); 
		for (int i = 1; i <= 30; i++) { 
			// 发送信息 
			String message = "你好；小兔子！work模式--" + i; 
			/**
			 * 参数1：交换机名称，如果没有指定则使用默认Default Exchage 
			 * 参数2：路由key,简单模式可以传递队列名称 
			 * 参数3：消息其它属性 
			 * 参数4：消息内容 
			*/ 
			channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); 
			System.out.println("已发送消息：" + message); 
		}
		// 关闭资源 
		channel.close(); connection.close(); 
	} 
}

package com.itheima.rabbitmq.work; 
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.*;
import java.io.IOException; 
public class Consumer1 { 
	public static void main(String[] args) throws Exception { 
		Connection connection = ConnectionUtil.getConnection(); 
		// 创建频道 
		Channel channel = connection.createChannel(); 
		// 声明（创建）队列 
		/**
		 * 参数1：队列名称 
		 * 参数2：是否定义持久化队列 
		 * 参数3：是否独占本次连接 
		 * 参数4：是否在不使用的时候自动删除队列 
		 * 参数5：队列其它参数 
		*/ 
		channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null); 
		//一次只能接收并处理一个消息 
		channel.basicQos(1); 
		//创建消费者；并设置消息处理 
		DefaultConsumer consumer = new DefaultConsumer(channel){ 
			@Override 
			/**
			 * consumerTag 消息者标签，在channel.basicConsume时候可以指定 
			 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) 
			 * properties 属性信息 
			 * body 消息 
			*/ 
			public void handleDelivery(String consumerTag, Envelope envelope, 
					AMQP.BasicProperties properties, byte[] body) throws IOException { 
				try {
					//路由key 
					System.out.println("路由key为：" + envelope.getRoutingKey()); 
					//交换机 
					System.out.println("交换机为：" + envelope.getExchange()); 
					//消息id 
					System.out.println("消息id为：" + envelope.getDeliveryTag()); 
					//收到的消息 
					System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8")); 
					Thread.sleep(1000); 
					//确认消息 
					channel.basicAck(envelope.getDeliveryTag(), false); 
				} 
				catch (InterruptedException e) { 
					e.printStackTrace(); 
				} 
			} 
		};
		//监听消息 
		/**
		 * 参数1：队列名称
		 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 
		 * 参数3：消息接收到后回调 
		*/ 
		channel.basicConsume(Producer.QUEUE_NAME, false, consumer); 
	} 
}

package com.itheima.rabbitmq.work; 
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.*; 
import java.io.IOException; 
public class Consumer2 { 
	public static void main(String[] args) throws Exception { 
		Connection connection = ConnectionUtil.getConnection(); 
		// 创建频道 
		Channel channel = connection.createChannel(); 
		// 声明（创建）队列 
		/**
		 * 参数1：队列名称 
		 * 参数2：是否定义持久化队列 
		 * 参数3：是否独占本次连接 
		 * 参数4：是否在不使用的时候自动删除队列 
		 * 参数5：队列其它参数 
		*/ 
		channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null); 
		//一次只能接收并处理一个消息 
		channel.basicQos(1); 
		//创建消费者；并设置消息处理 
		DefaultConsumer consumer = new DefaultConsumer(channel){ 
			@Override 
			/**
			 * consumerTag 消息者标签，在channel.basicConsume时候可以指定 
			 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) 
			 * properties 属性信息 
			 * body 消息 
			*/ 
			public void handleDelivery(String consumerTag, Envelope envelope, 
					AMQP.BasicProperties properties, byte[] body) throws IOException { 
				try {
					//路由key 
					System.out.println("路由key为：" + envelope.getRoutingKey()); 
					//交换机 
					System.out.println("交换机为：" + envelope.getExchange()); 
					//消息id 
					System.out.println("消息id为：" + envelope.getDeliveryTag());
					//收到的消息 
					System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8")); 
					Thread.sleep(1000); 
					//确认消息 
					channel.basicAck(envelope.getDeliveryTag(), false); 
				} catch (InterruptedException e) { 
					e.printStackTrace(); 
				} 
			} 
		};
		//监听消息 
		/**
		 * 参数1：队列名称 
		 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 
		 * 参数3：消息接收到后回调 
		*/ 
		channel.basicConsume(Producer.QUEUE_NAME, false, consumer); 
	} 
}