说明&＃xff1a;

&＃xff08;1&＃xff09;内容说明&＃xff1a;

          ● 这儿我们会创建一个项目&＃xff0c;演示RabbitMQ最基础的内容&＃xff1b;

通过&＃xff0c;这个最简单的例子&＃xff0c;先了解&＃xff1a;如何使用RabbitMQ&＃xff0c;如何连接RabbitMQ&＃xff0c;如何发送消息&＃xff0c;如何接收消息等最最基础的内容&＃xff1b;

          ● 然后&＃xff0c;会演示多个消费者平均压力的内容&＃xff1b;

目录

一&＃xff1a;第一个生产者和消费者&＃xff1b;

0.创建一个maven项目rabbitmq&＃xff0c;演示用&＃xff1b;

1.引入RabbitMQ的Java客户端的&＃xff0c;依赖&＃xff1b;

2.第一个生产者&＃xff1b;

3.第一个消费者&＃xff1b;

4.瞅一眼RabbitMQ管理后台&＃xff1b;

二&＃xff1a; 根据消息内容的不同&＃xff0c;采取不同的处理策略&＃xff1b;&＃xff08;这儿演示的是一种思路&＃xff09;

三&＃xff1a;当消息相对较多时&＃xff0c;多个消费者平均压力&＃xff1b; 

1.多个消费者&＃xff0c;平均压力&＃xff1a;引入&＃xff1b; 

2.公平派遣&＃xff1b;

一&＃xff1a;第一个生产者和消费者&＃xff1b;

0.创建一个maven项目rabbitmq&＃xff0c;演示用&＃xff1b;

 RabbitMQ支持多语言&＃xff0c;其中就包括Java&＃xff1b;同时&＃xff0c;RabbitMQ的API丰富&＃xff0c;我们可以利用RabbitMQ针对Java提供的客户端的一系列API&＃xff0c;来完成操作&＃xff1b;

1.引入RabbitMQ的Java客户端的&＃xff0c;依赖&＃xff1b;

com.rabbitmqamqp-client5.8.0org.slf4jslf4j-nop1.7.29

说明&＃xff1a;

&＃xff08;1&＃xff09;依赖说明&＃xff1b;

2.第一个生产者&＃xff1b;

Send类&＃xff1a;

package helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 描述&＃xff1a;发送消息的类&＃xff1a;连接到RabbitMQ的服务端&＃xff0c;然后发送一条消息&＃xff0c;然后退出&＃xff1b;*/
public class Send {//我们发送消息时&＃xff0c;需要指定要发到哪里去&＃xff1b;所以&＃xff0c;我们要指定队列的名字&＃xff1b;所以&＃xff0c;这儿我们定义队列的名字&＃xff1b;//这个名字可随便取&＃xff0c;待会在接收的消息时候&＃xff0c;要使用这个队列&＃xff1b;private final static String QUEUE_NAME &＃61; "hello";public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory connectionFactory &＃61; new ConnectionFactory();//2.设置RabbitMQ的地址&＃xff08;即RabbitMQ的服务端的地址&＃xff09;//这里面填写是RabbitMQ服务端所在服务器的ip地址connectionFactory.setHost("1**.***.***.**8");//然后&＃xff0c;要想连接RabbitMQ的服务端&＃xff0c;我么还需要通过一个用户才行&＃xff1b;// 所以&＃xff0c;这儿我们使用【前面我们设置的&＃xff0c;能够在其他服务器上访问RabbitMQ所在服务器的&＃xff0c;admin用户】connectionFactory.setUsername("admin");connectionFactory.setPassword("password");//PS:记得要放开RabbitMQ部署的服务器的&＃xff0c;5672端口&＃xff1b;//3.建立连接Connection connection &＃61; connectionFactory.newConnection();//4.获得Channel信道&＃xff08;我们大部分的操作&＃xff0c;都是在信道上完成的&＃xff1b;有了信道后&＃xff0c;我们就可以进行操作了&＃xff09;Channel channel &＃61; connection.createChannel();//5.声明队列&＃xff08;有了队列之后&＃xff0c;我们就可以发布消息了&＃xff09;//参数说明&＃xff1a;第一个参数(queue)&＃xff1a;队列名&＃xff1b;// 第二个参数(durable)&＃xff1a;这个队列是否需要持久&＃xff08;即&＃xff0c;服务重启后&＃xff0c;这个队列是否需要还存在&＃xff1b;这儿我们根据自己的需求&＃xff0c;设为了false&＃xff1b;&＃xff09;//第三个参数(exclusive)&＃xff1a;这个队列是否独有&＃xff08;即&＃xff0c;这个队列是不是仅能给这个连接使用&＃xff1b;这儿我们设为了false&＃xff09;//第四个参数(autoDelete)&＃xff1a;这个队列是否需要自动删除&＃xff08;即&＃xff0c;在队列没有使用的时候&＃xff0c;是否需要自动删除&＃xff1b;这儿我们设为了false&＃xff09;//第五个参数(arguments)&＃xff1b;channel.queueDeclare(QUEUE_NAME, false, false, false, null);//6.发布消息String message &＃61; "测试的消息";//参数说明&＃xff1a;第一个参数&＃xff08;exchange&＃xff09;是交换机&＃xff0c;这儿我们暂时不深入了解&＃xff1b;// 第二个参数&＃xff08;routingKey&＃xff09;是路由键&＃xff0c;这儿我们就写成队列的名字&＃xff1b;//第三个参数&＃xff08;props&＃xff09;&＃xff0c;消息除了消息体外&＃xff0c;还要有props作为它的配置&＃xff1b;// 第四个参数&＃xff08;body&＃xff09;消息的内容&＃xff0c;要求是byte[]类型的&＃xff0c;同时&＃xff0c;需要指定编码类型channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));System.out.println("消息发送成功了:" &＃43; message);//7.关闭连接&＃xff1a;先关闭channel信道&＃xff0c;然后关闭connection连接&＃xff1b;channel.close();connection.close();}}


说明&＃xff1a;

&＃xff08;0&＃xff09;RabbitMQ有三个所谓的“端”&＃xff1a;这儿梳理一下&＃xff1b;

          ● 服务端&＃xff1a;就是我们安装了RabbitMQ的Linux系统&＃xff1b;我们把RabbitMQ启动后&＃xff0c;这个服务器中的RabbitMQ就是服务端&＃xff1b;

          ● 管理后台&＃xff1a;RabbitMQ的管理后台&＃xff0c;就是我们在【RabbitMQ入门3&＃xff1a;RabbitMQ管理后台&＃xff0c;简介&＃xff1b;】中&＃xff0c;演示的在web端查看、管理RabbitMQ的一些配置的地方&＃xff1b;&＃xff08;PS&＃xff1a;要想在网页上访问管理后台&＃xff0c;那么部署RabbitMQ的服务器&＃xff0c;就要开发RabbitMQ的15672端口&＃xff09;

          ● 客户端&＃xff1a;比如&＃xff0c;在这儿&＃xff0c;我们在我们的Java项目中&＃xff0c;引入RabbitMQ提供的Java客户端后&＃xff0c;我们就可以通过客户端去操作RabbitMQ了&＃xff1b;&＃xff08;PS&＃xff1a;要想在远端服务器&＃xff0c;通过客户端访问RabbitMQ服务&＃xff0c;那么部署RabbitMQ的服务器&＃xff0c;就要开发RabbitMQ的5672端口&＃xff09;

&＃xff08;1&＃xff09;看注释&＃xff1b;这儿的连接RabbitMQ的套路&＃xff0c;都是相对固定的&＃xff0c;后面如有需要&＃xff0c;我们似乎也可以创建一个工具类&＃xff1b;

&＃xff08;2&＃xff09;我们要想在其他服务器&＃xff0c;通过客户端访问RabbitMQ&＃xff0c;那么部署RabbitMQ的服务器&＃xff0c;就要开发RabbitMQ的5672端口&＃xff1b;关于Linux防火墙设置&＃xff0c;可以参考【Linux进阶六&＃xff1a;【firewall-cmd】防火墙设置&＃xff1b;&＃xff08;以【对外开放Tomcat】为例来演示&＃xff09;】&＃xff1b;

&＃xff08;3&＃xff09;如果我们的RabbitMQ是安装在本机的话&＃xff0c;就可以设置本机地址&＃xff0c;然后其默认会使用guest用户去登录&＃xff0c;所以设置用户就可以省略&＃xff1b;

&＃xff08;4&＃xff09;运行结果&＃xff1a;

3.第一个消费者&＃xff1b;

Recv类&＃xff1a;

package helloworld;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 描述&＃xff1a;接收消息的类&＃xff1a;连接到RabbitMQ的服务端&＃xff0c;然后接收消息&＃xff1b;这个接收消息的类&＃xff0c;会持续运行&＃xff1b;*/
public class Recv {//我们这儿想要接收的消息&＃xff0c;就是Send类发送到“hello”这个队列中的消息&＃xff1b;// 所以&＃xff0c;在接收消息的时候&＃xff0c;我们也要使用到“hello”这个队列&＃xff1b;private final static String QUEUE_NAME &＃61; "hello";public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory connectionFactory &＃61; new ConnectionFactory();//2.设置RabbitMQ的地址&＃xff08;即RabbitMQ的服务端的地址&＃xff09;//这里面填写是RabbitMQ服务端所在服务器的ip地址connectionFactory.setHost("1**.***.***.**8");//然后&＃xff0c;要想连接RabbitMQ的服务端&＃xff0c;我么还需要通过一个用户才行&＃xff1b;// 所以&＃xff0c;这儿我们使用【前面我们设置的&＃xff0c;能够在其他服务器上访问RabbitMQ所在服务器的&＃xff0c;admin用户】connectionFactory.setUsername("admin");connectionFactory.setPassword("password");//PS:记得要放开RabbitMQ部署的服务器的&＃xff0c;5672端口&＃xff1b;//3.建立连接Connection connection &＃61; connectionFactory.newConnection();//4.获得Channel信道&＃xff08;我们大部分的操作&＃xff0c;都是在信道上完成的&＃xff1b;有了信道后&＃xff0c;我们就可以进行操作了&＃xff09;Channel channel &＃61; connection.createChannel();//5.声明队列&＃xff1a;因为这儿想要接收的消息&＃xff0c;就是Send类发送到“hello”这个队列中的消息&＃xff1b;所以&＃xff0c;这儿声明的队列和Send中声明的队列是一样的&＃xff1b;//参数说明&＃xff1a;第一个参数(queue)&＃xff1a;队列名&＃xff1b;// 第二个参数(durable)&＃xff1a;这个队列是否需要持久&＃xff08;即&＃xff0c;服务重启后&＃xff0c;这个队列是否需要还存在&＃xff1b;这儿我们根据自己的需求&＃xff0c;设为了false&＃xff1b;&＃xff09;//第三个参数(exclusive)&＃xff1a;这个队列是否独有&＃xff08;即&＃xff0c;这个队列是不是仅能给这个连接使用&＃xff1b;这儿我们设为了false&＃xff09;//第四个参数(autoDelete)&＃xff1a;这个队列是否需要自动删除&＃xff08;即&＃xff0c;在队列没有使用的时候&＃xff0c;是否需要自动删除&＃xff1b;这儿我们设为了false&＃xff09;//第五个参数(arguments)&＃xff1b;channel.queueDeclare(QUEUE_NAME, false, false, false, null);//6.接收消息&＃xff0c;并消费//参数说明&＃xff1a;第一个参数&＃xff08;queue&＃xff09;队列名&＃xff1b;// 第二个参数&＃xff08;autoAck&＃xff09;&＃xff0c;是否去自动的确认收到&＃xff1b;即&＃xff0c;这儿接收到消息之后&＃xff0c;是否需要通知消息发送者&＃xff1b;//第三个参数&＃xff08;callback&＃xff09;&＃xff0c;消息收到后的处理channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel){&＃64;Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message &＃61; new String(body, "UTF-8");System.out.println(message);}});}
}


说明&＃xff1a;

&＃xff08;1&＃xff09;接收消息的时候&＃xff0c;我们不需要关闭channel和connection&＃xff0c;这个类会一直运行&＃xff0c;即其会一直看队列中有没有数据&＃xff0c;有的话就拿过来消费&＃xff1b;

&＃xff08;2&＃xff09;看注释&＃xff1b;

&＃xff08;3&＃xff09;我们接收到消息后&＃xff0c;具体消息怎么处理&＃xff0c;写在了handleDelivery()方法中&＃xff1b;

&＃xff08;4&＃xff09;运行结果&＃xff1b;

4.瞅一眼RabbitMQ管理后台&＃xff1b;

二&＃xff1a; 根据消息内容的不同&＃xff0c;采取不同的处理策略&＃xff1b;&＃xff08;这儿演示的是一种思路&＃xff09;

在实际开发中&＃xff0c;有时可能需要根据消息内容的不同&＃xff0c;采取不同的处理策略&＃xff1b;本篇博客&＃xff0c;就来演示一下&＃xff1b;

NewTask类&＃xff1a;

package workqueues;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 描述&＃xff1a;发送消息的类&＃xff1a;其会发送*/
public class NewTask {//我们发送消息时&＃xff0c;需要指定要发到哪里去&＃xff1b;所以&＃xff0c;我们要指定队列的名字&＃xff1b;所以&＃xff0c;这儿我们定义队列的名字&＃xff1b;//这个名字可随便取&＃xff0c;待会在接收的消息时候&＃xff0c;要使用这个队列&＃xff1b;private final static String TASK_QUEUE_NAME &＃61; "task_queue";public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory connectionFactory &＃61; new ConnectionFactory();//2.设置RabbitMQ的地址&＃xff08;即RabbitMQ的服务端的地址&＃xff09;//这里面填写是RabbitMQ服务端所在服务器的ip地址connectionFactory.setHost("1**.***.***.**8");//然后&＃xff0c;要想连接RabbitMQ的服务端&＃xff0c;我么还需要通过一个用户才行&＃xff1b;// 所以&＃xff0c;这儿我们使用【前面我们设置的&＃xff0c;能够在其他服务器上访问RabbitMQ所在服务器的&＃xff0c;admin用户】connectionFactory.setUsername("admin");connectionFactory.setPassword("password");//PS:记得要放开RabbitMQ部署的服务器的&＃xff0c;5672端口&＃xff1b;//3.建立连接Connection connection &＃61; connectionFactory.newConnection();//4.获得Channel信道&＃xff08;我们大部分的操作&＃xff0c;都是在信道上完成的&＃xff1b;有了信道后&＃xff0c;我们就可以进行操作了&＃xff09;Channel channel &＃61; connection.createChannel();//5.声明队列&＃xff08;有了队列之后&＃xff0c;我们就可以发布消息了&＃xff09;//参数说明&＃xff1a;第一个参数(queue)&＃xff1a;队列名&＃xff1b;// 第二个参数(durable)&＃xff1a;这个队列是否需要持久&＃xff08;即&＃xff0c;服务重启后&＃xff0c;这个队列是否需要还存在&＃xff1b;这儿我们根据自己的需求&＃xff0c;设为了true&＃xff1b;&＃xff09;//第三个参数(exclusive)&＃xff1a;这个队列是否独有&＃xff08;即&＃xff0c;这个队列是不是仅能给这个连接使用&＃xff1b;这儿我们设为了false&＃xff09;//第四个参数(autoDelete)&＃xff1a;这个队列是否需要自动删除&＃xff08;即&＃xff0c;在队列没有使用的时候&＃xff0c;是否需要自动删除&＃xff1b;这儿我们设为了false&＃xff09;//第五个参数(arguments)&＃xff1b;channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);//6.发布消息;这儿我们模拟发送10条消息&＃xff08;消息内容是&＃xff0c;如"1...","2...","3..."……这样的&＃xff09;for (int i &＃61; 0; i <10; i&＃43;&＃43;) {String message &＃61; i &＃43; "...";//参数说明&＃xff1a;第一个参数&＃xff08;exchange&＃xff09;是交换机&＃xff0c;这儿我们暂时不深入了解&＃xff1b;// 第二个参数&＃xff08;routingKey&＃xff09;是路由键&＃xff0c;这儿我们就写成队列的名字&＃xff1b;//第三个参数&＃xff08;props&＃xff09;&＃xff0c;消息除了消息体外&＃xff0c;还要有props作为它的配置&＃xff1b;// 第四个参数&＃xff08;body&＃xff09;消息的内容&＃xff0c;要求是byte[]类型的&＃xff0c;同时&＃xff0c;需要指定编码类型channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));System.out.println("消息发送成功了:" &＃43; message);}//7.关闭连接&＃xff1a;先关闭channel信道&＃xff0c;然后关闭connection连接&＃xff1b;channel.close();connection.close();}
}


说明&＃xff1a;

&＃xff08;1&＃xff09;看注释&＃xff1b;

&＃xff08;2&＃xff09;类内容说明&＃xff1b;

Worker类&＃xff1a;

package workqueues;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 描述&＃xff1a;接收消息的类&＃xff1a;连接到RabbitMQ的服务端&＃xff0c;然后接收消息&＃xff1b;这个接收消息的类&＃xff0c;会持续运行&＃xff1b;*/
public class Worker {//我们这儿想要接收的消息&＃xff0c;就是NewTask类发送到“task_queue”这个队列中的消息&＃xff1b;// 所以&＃xff0c;在接收消息的时候&＃xff0c;我们也要使用到“task_queue”这个队列&＃xff1b;private final static String TASK_QUEUE_NAME &＃61; "task_queue";public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory connectionFactory &＃61; new ConnectionFactory();//2.设置RabbitMQ的地址&＃xff08;即RabbitMQ的服务端的地址&＃xff09;//这里面填写是RabbitMQ服务端所在服务器的ip地址connectionFactory.setHost("1**.***.***.**8");//然后&＃xff0c;要想连接RabbitMQ的服务端&＃xff0c;我么还需要通过一个用户才行&＃xff1b;// 所以&＃xff0c;这儿我们使用【前面我们设置的&＃xff0c;能够在其他服务器上访问RabbitMQ所在服务器的&＃xff0c;admin用户】connectionFactory.setUsername("admin");connectionFactory.setPassword("password");//PS:记得要放开RabbitMQ部署的服务器的&＃xff0c;5672端口&＃xff1b;//3.建立连接Connection connection &＃61; connectionFactory.newConnection();//4.获得Channel信道&＃xff08;我们大部分的操作&＃xff0c;都是在信道上完成的&＃xff1b;有了信道后&＃xff0c;我们就可以进行操作了&＃xff09;Channel channel &＃61; connection.createChannel();//5.声明队列&＃xff1a;因为这儿想要接收的消息&＃xff0c;就是Send类发送到“hello”这个队列中的消息&＃xff1b;所以&＃xff0c;这儿声明的队列和Send中声明的队列是一样的&＃xff1b;//参数说明&＃xff1a;第一个参数(queue)&＃xff1a;队列名&＃xff1b;// 第二个参数(durable)&＃xff1a;这个队列是否需要持久&＃xff08;即&＃xff0c;服务重启后&＃xff0c;这个队列是否需要还存在&＃xff1b;这儿我们根据自己的需求&＃xff0c;设为了true&＃xff1b;&＃xff09;//第三个参数(exclusive)&＃xff1a;这个队列是否独有&＃xff08;即&＃xff0c;这个队列是不是仅能给这个连接使用&＃xff1b;这儿我们设为了false&＃xff09;//第四个参数(autoDelete)&＃xff1a;这个队列是否需要自动删除&＃xff08;即&＃xff0c;在队列没有使用的时候&＃xff0c;是否需要自动删除&＃xff1b;这儿我们设为了false&＃xff09;//第五个参数(arguments)&＃xff1b;channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);System.out.println("开始接收消息");//6.接收消息&＃xff0c;并消费//参数说明&＃xff1a;第一个参数&＃xff08;queue&＃xff09;队列名&＃xff1b;// 第二个参数&＃xff08;autoAck&＃xff09;&＃xff0c;是否去自动的确认收到&＃xff1b;即&＃xff0c;这儿接收到消息之后&＃xff0c;是否需要通知消息发送者&＃xff1b;//第三个参数&＃xff08;callback&＃xff09;&＃xff0c;消息收到后的处理channel.basicConsume(TASK_QUEUE_NAME, true, new DefaultConsumer(channel){&＃64;Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message &＃61; new String(body, "UTF-8");System.out.println("收到了消息" &＃43; message);try{doWork(message);}finally {System.out.println("完成消息处理");}}});}/*** 工具方法&＃xff1a;处理消息&＃xff1b;* &＃64;param task*/private static void doWork(String task) {//根据具体消息内容的不同&＃xff0c;去处理消息&＃xff1b;// 即&＃xff0c;如果消息中有&＃39;.&＃39;的话&＃xff0c;那么我们就让其处理速度慢1秒&＃xff1b;&＃xff08;PS&＃xff1a;这儿仅仅是为了演示用的&＃xff0c;玩具式程序&＃xff09;// 那么&＃xff0c;这样一来&＃xff0c;就会出现这个效果&＃xff1a;如果消息中没有&＃39;.&＃39;&＃xff0c;处理的就会很快&＃xff1b;如果有&＃39;.&＃39;&＃xff0c;处理速度就会慢的多&＃xff1b;char[] chars &＃61; task.toCharArray();for (char ch : chars) {if (ch &＃61;&＃61; &＃39;.&＃39;) {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}}
}


说明&＃xff1a;

&＃xff08;1&＃xff09;看注释&＃xff1b;

&＃xff08;2&＃xff09;类内容说明&＃xff1b;

&＃xff08;3&＃xff09;运行结果&＃xff1b;

---

三&＃xff1a;当消息相对较多时&＃xff0c;多个消费者平均压力&＃xff1b; 

可以看到&＃xff0c;在&＃xff08;二&＃xff1a; 根据消息内容的不同&＃xff0c;采取不同的处理策略&＃xff1b;&＃xff08;这儿演示的是一种思路&＃xff09;&＃xff09;中&＃xff0c;单靠Worker处理10条信息&＃xff0c;需要10秒多&＃xff1b;试想&＃xff0c;我们再创建一个Worker&＃xff0c;10条消息的处理速度就会提升&＃xff1b;

1.多个消费者&＃xff0c;平均压力&＃xff1a;引入&＃xff1b; 

一个直接的想法就是&＃xff0c;既然我们想让多个Worker一起来接收并处理消息&＃xff0c;那么我们可不可以再运行一次Worker类&＃xff1f;

进行一下配置&＃xff0c;让Worker类可以有多个实例并行运行&＃xff1b;

那么&＃xff0c;此时的效果&＃xff1b;

默认情况下&＃xff0c;如果有多个Worker的话&＃xff0c;那么这多个Worker会并行工作&＃xff1b;RabbitMQ会根据已启动worker和消息的情况&＃xff0c;按顺序把每个消息发送给下一个Worker&＃xff0c;在消息数量上是平均分配的&＃xff1b;即&＃xff0c;比如上面有10条消息&＃xff0c;两个消费者&＃xff0c;那么无论两个消费者处理能力如何&＃xff0c;每个消费者都会收到5条消息&＃xff1b;

但是&＃xff0c;这种平均&＃xff0c;是任务量的平均分配&＃xff0c;而不一定是真实工作量&＃xff08;压力&＃xff09;的分配&＃xff1b;比如&＃xff0c;下面的案例&＃xff1b;

此时&＃xff0c;再发送消息&＃xff0c;观察效果&＃xff1a;

而&＃xff0c;为了解决这种&＃xff0c;纯按数量和顺序分配&＃xff0c;却没有按工作量&＃xff08;压力&＃xff09;平均分配的问题&＃xff1b;就是下面公平派遣的内容了&＃xff1b;

2.公平派遣&＃xff1b;

公平派遣是在有多个消费者&＃xff0c;而且是循环调度的情况下&＃xff0c;来说的&＃xff1b;公平派遣机制下&＃xff0c;RabbitMQ会根据消费者的压力&＃xff0c;来决定是否派遣&＃xff1b;

要想实现公平派遣&＃xff0c;还需要加入消息确认机制&＃xff1b;主要目的是&＃xff0c;消费者处理完消息后&＃xff0c;发送一个确认消息&＃xff0c;这样一来&＃xff0c;RabbitMQ就知道你处理完了&＃xff0c;然后就会给你发送下一个消息&＃xff1b;

修改Worker这个Consumer类的内容如下&＃xff1a;

package workqueues;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 描述&＃xff1a;接收消息的类&＃xff1a;连接到RabbitMQ的服务端&＃xff0c;然后接收消息&＃xff1b;这个接收消息的类&＃xff0c;会持续运行&＃xff1b;*/
public class Worker {//我们这儿想要接收的消息&＃xff0c;就是NewTask类发送到“task_queue”这个队列中的消息&＃xff1b;// 所以&＃xff0c;在接收消息的时候&＃xff0c;我们也要使用到“task_queue”这个队列&＃xff1b;private final static String TASK_QUEUE_NAME &＃61; "task_queue";public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory connectionFactory &＃61; new ConnectionFactory();//2.设置RabbitMQ的地址&＃xff08;即RabbitMQ的服务端的地址&＃xff09;//这里面填写是RabbitMQ服务端所在服务器的ip地址connectionFactory.setHost("1**.***.***.**8");//然后&＃xff0c;要想连接RabbitMQ的服务端&＃xff0c;我么还需要通过一个用户才行&＃xff1b;// 所以&＃xff0c;这儿我们使用【前面我们设置的&＃xff0c;能够在其他服务器上访问RabbitMQ所在服务器的&＃xff0c;admin用户】connectionFactory.setUsername("admin");connectionFactory.setPassword("password");//PS:记得要放开RabbitMQ部署的服务器的&＃xff0c;5672端口&＃xff1b;//3.建立连接Connection connection &＃61; connectionFactory.newConnection();//4.获得Channel信道&＃xff08;我们大部分的操作&＃xff0c;都是在信道上完成的&＃xff1b;有了信道后&＃xff0c;我们就可以进行操作了&＃xff09;final Channel channel &＃61; connection.createChannel();//5.声明队列&＃xff1a;因为这儿想要接收的消息&＃xff0c;就是Send类发送到“hello”这个队列中的消息&＃xff1b;所以&＃xff0c;这儿声明的队列和Send中声明的队列是一样的&＃xff1b;//参数说明&＃xff1a;第一个参数(queue)&＃xff1a;队列名&＃xff1b;// 第二个参数(durable)&＃xff1a;这个队列是否需要持久&＃xff08;即&＃xff0c;服务重启后&＃xff0c;这个队列是否需要还存在&＃xff1b;这儿我们根据自己的需求&＃xff0c;设为了true&＃xff1b;&＃xff09;//第三个参数(exclusive)&＃xff1a;这个队列是否独有&＃xff08;即&＃xff0c;这个队列是不是仅能给这个连接使用&＃xff1b;这儿我们设为了false&＃xff09;//第四个参数(autoDelete)&＃xff1a;这个队列是否需要自动删除&＃xff08;即&＃xff0c;在队列没有使用的时候&＃xff0c;是否需要自动删除&＃xff1b;这儿我们设为了false&＃xff09;//第五个参数(arguments)&＃xff1b;channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);System.out.println("开始接收消息");//这句话的意思&＃xff0c;这个消费者最希望处理的消息的数量&＃xff1b;那么效果就是&＃xff0c;这个消费者在处理完一个消息之前&＃xff0c;是不会接收下一个消息的&＃xff1b;channel.basicQos(1);//6.接收消息&＃xff0c;并消费//参数说明&＃xff1a;第一个参数&＃xff08;queue&＃xff09;队列名&＃xff1b;// 第二个参数&＃xff08;autoAck&＃xff09;&＃xff0c;是否去自动的确认收到&＃xff1b;即&＃xff0c;这儿接收到消息之后&＃xff0c;是否需要通知消息发送者&＃xff1b;//为了演示公平派遣&＃xff0c;我们这儿改成了false&＃xff0c;即我么手动却发送处理了完成的消息//第三个参数&＃xff08;callback&＃xff09;&＃xff0c;消息收到后的处理channel.basicConsume(TASK_QUEUE_NAME, false, new DefaultConsumer(channel){&＃64;Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message &＃61; new String(body, "UTF-8");System.out.println("收到了消息" &＃43; message);try{doWork(message);}finally {System.out.println("完成消息处理");//消息处理完成后&＃xff0c;去手动确认&＃xff1b;//第一个参数(deliveryTag)这个参数是固定的&＃xff1b;//第二参数(multiple)意思是&＃xff0c;我们是否同时多个消息一起确认&＃xff0c;这儿我们不需要&＃xff0c;所以设为了false&＃xff1b;channel.basicAck(envelope.getDeliveryTag(), false);}}});}/*** 工具方法&＃xff1a;处理消息&＃xff1b;* &＃64;param task*/private static void doWork(String task) {//根据具体消息内容的不同&＃xff0c;去处理消息&＃xff1b;// 即&＃xff0c;如果消息中有&＃39;.&＃39;的话&＃xff0c;那么我们就让其处理速度慢1秒&＃xff1b;&＃xff08;PS&＃xff1a;这儿仅仅是为了演示用的&＃xff0c;玩具式程序&＃xff09;// 那么&＃xff0c;这样一来&＃xff0c;就会出现这个效果&＃xff1a;如果消息中没有&＃39;.&＃39;&＃xff0c;处理的就会很快&＃xff1b;如果有&＃39;.&＃39;&＃xff0c;处理速度就会慢的多&＃xff1b;char[] chars &＃61; task.toCharArray();for (char ch : chars) {if (ch &＃61;&＃61; &＃39;.&＃39;) {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}}
}


说明&＃xff1a;

&＃xff08;1&＃xff09;修改内容说明&＃xff1b;

&＃xff08;2&＃xff09;注意&＃xff1a;此时&＃xff0c;一定要记得主动发送【消息处理完成的通知】&＃xff1b;否则&＃xff0c;RabbitMQ就不知道&＃xff0c;这个消息是否被处理完了&＃xff0c;其就会认为没有被处理完&＃xff0c;于是后续的消息就会得不到处理&＃xff0c;越积越多&＃xff0c;消耗内存&＃xff1b;

&＃xff08;3&＃xff09;运行效果&＃xff1b;此时&＃xff0c;我们重开两个Worker&＃xff0c;然后运行NewTask&＃xff0c;发送10条消息&＃xff1b;