up:: RabbitMQ管理后台简介
说明：
（1） 内容说明：
● 这儿我们会创建一个项目，演示RabbitMQ最基础的内容；
通过，这个最简单的例子，先了解：如何使用RabbitMQ，如何连接RabbitMQ，如何发送消息，如何接收消息等最最基础的内容；
● 然后，会演示多个消费者平均压力的内容；
一：第一个生产者和消费者；
0.创建一个maven项目rabbitmq，演示用；
RabbitMQ支持多语言，其中就包括Java；同时，RabbitMQ的API丰富，我们可以利用RabbitMQ针对Java提供的客户端的一系列API，来完成操作；
1.引入RabbitMQ的Java客户端的，依赖；
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.8.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-nop</artifactId>
<version>1.7.29</version>
</dependency>
</dependencies>说明：
（1） 依赖说明；
2.第一个生产者；
Send类：
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;
/**
* 描述：发送消息的类：连接到RabbitMQ的服务端，然后发送一条消息，然后退出；
*/
public class Send {
//我们发送消息时，需要指定要发到哪里去；所以，我们要指定队列的名字；所以，这儿我们定义队列的名字；
//这个名字可随便取，待会在接收的消息时候，要使用这个队列；
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//2.设置RabbitMQ的地址（即RabbitMQ的服务端的地址）
//这里面填写是RabbitMQ服务端所在服务器的ip地址
connectionFactory.setHost("1**.***.***.**8");
//然后，要想连接RabbitMQ的服务端，我么还需要通过一个用户才行；
// 所以，这儿我们使用【前面我们设置的，能够在其他服务器上访问RabbitMQ所在服务器的，admin用户】
connectionFactory.setUsername("admin");
connectionFactory.setPassword("password");
//PS:记得要放开RabbitMQ部署的服务器的，5672端口；
//3.建立连接
Connection connection = connectionFactory.newConnection();
//4.获得Channel信道（我们大部分的操作，都是在信道上完成的；有了信道后，我们就可以进行操作了）
Channel channel = connection.createChannel();
//5.声明队列（有了队列之后，我们就可以发布消息了）
//参数说明：第一个参数(queue)：队列名；
// 第二个参数(durable)：这个队列是否需要持久（即，服务重启后，这个队列是否需要还存在；这儿我们根据自己的需求，设为了false；）
//第三个参数(exclusive)：这个队列是否独有（即，这个队列是不是仅能给这个连接使用；这儿我们设为了false）
//第四个参数(autoDelete)：这个队列是否需要自动删除（即，在队列没有使用的时候，是否需要自动删除；这儿我们设为了false）
//第五个参数(arguments)；
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//6.发布消息
String message = "测试的消息";
//参数说明：第一个参数（exchange）是交换机，这儿我们暂时不深入了解；
// 第二个参数（routingKey）是路由键，这儿我们就写成队列的名字；
//第三个参数（props），消息除了消息体外，还要有props作为它的配置；
// 第四个参数（body）消息的内容，要求是byte[]类型的，同时，需要指定编码类型
channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));
System.out.println("消息发送成功了:" + message);
//7.关闭连接：先关闭channel信道，然后关闭connection连接；
channel.close();
connection.close();
}
}
说明：
（0） RabbitMQ有三个所谓的“端”：这儿梳理一下；
● 服务端：就是我们安装了RabbitMQ的Linux系统；我们把RabbitMQ启动后，这个服务器中的RabbitMQ就是服务端；
● 管理后台：RabbitMQ的管理后台，就是我们在【RabbitMQ管理后台简介】中，演示的在web端查看、管理RabbitMQ的一些配置的地方；（PS：要想在网页上访问管理后台，那么部署RabbitMQ的服务器，就要开发RabbitMQ的15672端口）
● 客户端：比如，在这儿，我们在我们的Java项目中，引入RabbitMQ提供的Java客户端后，我们就可以通过客户端去操作RabbitMQ了；（PS：要想在远端服务器，通过客户端访问RabbitMQ服务，那么部署RabbitMQ的服务器，就要开发RabbitMQ的5672端口）
（1） 看注释；这儿的连接RabbitMQ的套路，都是相对固定的，后面如有需要，我们似乎也可以创建一个工具类；
（2） 我们要想在其他服务器，通过客户端访问RabbitMQ，那么部署RabbitMQ的服务器，就要开发RabbitMQ的5672端口；关于Linux防火墙设置，可以参考【Linux环境搭建与Redis应用(三)】；
（3） 如果我们的RabbitMQ是安装在本机的话，就可以设置本机地址，然后其默认会使用guest用户去登录，所以设置用户就可以省略；
（4） 运行结果：
3.第一个消费者；
Recv类：
package helloworld;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 描述：接收消息的类：连接到RabbitMQ的服务端，然后接收消息；这个接收消息的类，会持续运行；
*/
public class Recv {
//我们这儿想要接收的消息，就是Send类发送到“hello”这个队列中的消息；
// 所以，在接收消息的时候，我们也要使用到“hello”这个队列；
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//2.设置RabbitMQ的地址（即RabbitMQ的服务端的地址）
//这里面填写是RabbitMQ服务端所在服务器的ip地址
connectionFactory.setHost("1**.***.***.**8");
//然后，要想连接RabbitMQ的服务端，我么还需要通过一个用户才行；
// 所以，这儿我们使用【前面我们设置的，能够在其他服务器上访问RabbitMQ所在服务器的，admin用户】
connectionFactory.setUsername("admin");
connectionFactory.setPassword("password");
//PS:记得要放开RabbitMQ部署的服务器的，5672端口；
//3.建立连接
Connection connection = connectionFactory.newConnection();
//4.获得Channel信道（我们大部分的操作，都是在信道上完成的；有了信道后，我们就可以进行操作了）
Channel channel = connection.createChannel();
//5.声明队列：因为这儿想要接收的消息，就是Send类发送到“hello”这个队列中的消息；所以，这儿声明的队列和Send中声明的队列是一样的；
//参数说明：第一个参数(queue)：队列名；
// 第二个参数(durable)：这个队列是否需要持久（即，服务重启后，这个队列是否需要还存在；这儿我们根据自己的需求，设为了false；）
//第三个参数(exclusive)：这个队列是否独有（即，这个队列是不是仅能给这个连接使用；这儿我们设为了false）
//第四个参数(autoDelete)：这个队列是否需要自动删除（即，在队列没有使用的时候，是否需要自动删除；这儿我们设为了false）
//第五个参数(arguments)；
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//6.接收消息，并消费
//参数说明：第一个参数（queue）队列名；
// 第二个参数（autoAck），是否去自动的确认收到；即，这儿接收到消息之后，是否需要通知消息发送者；
//第三个参数（callback），消息收到后的处理
channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(message);
}
});
}
}说明：
（1） 接收消息的时候，我们不需要关闭channel和connection，这个类会一直运行，即其会一直看队列中有没有数据，有的话就拿过来消费；
（2） 看注释；
（3） 我们接收到消息后，具体消息怎么处理，写在了handleDelivery()方法中；
（4） 运行结果；
4.瞅一眼RabbitMQ管理后台；
二： 根据消息内容的不同，采取不同的处理策略；（这儿演示的是一种思路）
在实际开发中，有时可能需要根据消息内容的不同，采取不同的处理策略；本篇博客，就来演示一下；
NewTask类：
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;
/**
* 描述：发送消息的类：其会发送
*/
public class NewTask {
//我们发送消息时，需要指定要发到哪里去；所以，我们要指定队列的名字；所以，这儿我们定义队列的名字；
//这个名字可随便取，待会在接收的消息时候，要使用这个队列；
private final static String TASK_QUEUE_NAME = "task_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//2.设置RabbitMQ的地址（即RabbitMQ的服务端的地址）
//这里面填写是RabbitMQ服务端所在服务器的ip地址
connectionFactory.setHost("1**.***.***.**8");
//然后，要想连接RabbitMQ的服务端，我么还需要通过一个用户才行；
// 所以，这儿我们使用【前面我们设置的，能够在其他服务器上访问RabbitMQ所在服务器的，admin用户】
connectionFactory.setUsername("admin");
connectionFactory.setPassword("password");
//PS:记得要放开RabbitMQ部署的服务器的，5672端口；
//3.建立连接
Connection connection = connectionFactory.newConnection();
//4.获得Channel信道（我们大部分的操作，都是在信道上完成的；有了信道后，我们就可以进行操作了）
Channel channel = connection.createChannel();
//5.声明队列（有了队列之后，我们就可以发布消息了）
//参数说明：第一个参数(queue)：队列名；
// 第二个参数(durable)：这个队列是否需要持久（即，服务重启后，这个队列是否需要还存在；这儿我们根据自己的需求，设为了true；）
//第三个参数(exclusive)：这个队列是否独有（即，这个队列是不是仅能给这个连接使用；这儿我们设为了false）
//第四个参数(autoDelete)：这个队列是否需要自动删除（即，在队列没有使用的时候，是否需要自动删除；这儿我们设为了false）
//第五个参数(arguments)；
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
//6.发布消息;这儿我们模拟发送10条消息（消息内容是，如"1...","2...","3..."……这样的）
for (int i = 0; i < 10; i++) {
String message = i + "...";
//参数说明：第一个参数（exchange）是交换机，这儿我们暂时不深入了解；
// 第二个参数（routingKey）是路由键，这儿我们就写成队列的名字；
//第三个参数（props），消息除了消息体外，还要有props作为它的配置；
// 第四个参数（body）消息的内容，要求是byte[]类型的，同时，需要指定编码类型
channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));
System.out.println("消息发送成功了:" + message);
}
//7.关闭连接：先关闭channel信道，然后关闭connection连接；
channel.close();
connection.close();
}
}
说明：
（1） 看注释；
（2） 类内容说明；
Worker类：
package workqueues;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 描述：接收消息的类：连接到RabbitMQ的服务端，然后接收消息；这个接收消息的类，会持续运行；
*/
public class Worker {
//我们这儿想要接收的消息，就是NewTask类发送到“task_queue”这个队列中的消息；
// 所以，在接收消息的时候，我们也要使用到“task_queue”这个队列；
private final static String TASK_QUEUE_NAME = "task_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//2.设置RabbitMQ的地址（即RabbitMQ的服务端的地址）
//这里面填写是RabbitMQ服务端所在服务器的ip地址
connectionFactory.setHost("1**.***.***.**8");
//然后，要想连接RabbitMQ的服务端，我么还需要通过一个用户才行；
// 所以，这儿我们使用【前面我们设置的，能够在其他服务器上访问RabbitMQ所在服务器的，admin用户】
connectionFactory.setUsername("admin");
connectionFactory.setPassword("password");
//PS:记得要放开RabbitMQ部署的服务器的，5672端口；
//3.建立连接
Connection connection = connectionFactory.newConnection();
//4.获得Channel信道（我们大部分的操作，都是在信道上完成的；有了信道后，我们就可以进行操作了）
Channel channel = connection.createChannel();
//5.声明队列：因为这儿想要接收的消息，就是Send类发送到“hello”这个队列中的消息；所以，这儿声明的队列和Send中声明的队列是一样的；
//参数说明：第一个参数(queue)：队列名；
// 第二个参数(durable)：这个队列是否需要持久（即，服务重启后，这个队列是否需要还存在；这儿我们根据自己的需求，设为了true；）
//第三个参数(exclusive)：这个队列是否独有（即，这个队列是不是仅能给这个连接使用；这儿我们设为了false）
//第四个参数(autoDelete)：这个队列是否需要自动删除（即，在队列没有使用的时候，是否需要自动删除；这儿我们设为了false）
//第五个参数(arguments)；
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
System.out.println("开始接收消息");
//6.接收消息，并消费
//参数说明：第一个参数（queue）队列名；
// 第二个参数（autoAck），是否去自动的确认收到；即，这儿接收到消息之后，是否需要通知消息发送者；
//第三个参数（callback），消息收到后的处理
channel.basicConsume(TASK_QUEUE_NAME, true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("收到了消息" + message);
try{
doWork(message);
}finally {
System.out.println("完成消息处理");
}
}
});
}
/**
* 工具方法：处理消息；
* @param task
*/
private static void doWork(String task) {
//根据具体消息内容的不同，去处理消息；
// 即，如果消息中有'.'的话，那么我们就让其处理速度慢1秒；（PS：这儿仅仅是为了演示用的，玩具式程序）
// 那么，这样一来，就会出现这个效果：如果消息中没有'.'，处理的就会很快；如果有'.'，处理速度就会慢的多；
char[] chars = task.toCharArray();
for (char ch : chars) {
if (ch == '.') {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}说明：
（1） 看注释；
（2） 类内容说明；
（3） 运行结果；
三：当消息相对较多时，多个消费者平均压力；
可以看到，在（二：根据消息内容的不同，采取不同的处理策略；（这儿演示的是一种思路））中，单靠Worker处理10条信息，需要10秒多；试想，我们再创建一个Worker，10条消息的处理速度就会提升；
1.多个消费者，平均压力：引入；
一个直接的想法就是，既然我们想让多个Worker一起来接收并处理消息，那么我们可不可以再运行一次Worker类？
进行一下配置，让Worker类可以有多个实例并行运行；
那么，此时的效果；
默认情况下，如果有多个Worker的话，那么这多个Worker会并行工作；RabbitMQ会根据已启动worker和消息的情况，按顺序把每个消息发送给下一个Worker，在消息数量上是平均分配的；即，比如上面有10条消息，两个消费者，那么无论两个消费者处理能力如何，每个消费者都会收到5条消息；> 但是，这种平均，是任务量的平均分配，而不一定是真实工作量（压力）的分配；比如，下面的案例；
此时，再发送消息，观察效果：
而，为了解决这种，纯按数量和顺序分配，却没有按工作量（压力）平均分配的问题；就是下面公平派遣的内容了；
2.公平派遣；
公平派遣是在有多个消费者，而且是循环调度的情况下，来说的；公平派遣机制下，RabbitMQ会根据消费者的压力，来决定是否派遣；
要想实现公平派遣，还需要加入消息确认机制；主要目的是，消费者处理完消息后，发送一个确认消息，这样一来，RabbitMQ就知道你处理完了，然后就会给你发送下一个消息；
修改Worker这个Consumer类的内容如下：
package workqueues;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 描述：接收消息的类：连接到RabbitMQ的服务端，然后接收消息；这个接收消息的类，会持续运行；
*/
public class Worker {
//我们这儿想要接收的消息，就是NewTask类发送到“task_queue”这个队列中的消息；
// 所以，在接收消息的时候，我们也要使用到“task_queue”这个队列；
private final static String TASK_QUEUE_NAME = "task_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//2.设置RabbitMQ的地址（即RabbitMQ的服务端的地址）
//这里面填写是RabbitMQ服务端所在服务器的ip地址
connectionFactory.setHost("1**.***.***.**8");
//然后，要想连接RabbitMQ的服务端，我么还需要通过一个用户才行；
// 所以，这儿我们使用【前面我们设置的，能够在其他服务器上访问RabbitMQ所在服务器的，admin用户】
connectionFactory.setUsername("admin");
connectionFactory.setPassword("password");
//PS:记得要放开RabbitMQ部署的服务器的，5672端口；
//3.建立连接
Connection connection = connectionFactory.newConnection();
//4.获得Channel信道（我们大部分的操作，都是在信道上完成的；有了信道后，我们就可以进行操作了）
final Channel channel = connection.createChannel();
//5.声明队列：因为这儿想要接收的消息，就是Send类发送到“hello”这个队列中的消息；所以，这儿声明的队列和Send中声明的队列是一样的；
//参数说明：第一个参数(queue)：队列名；
// 第二个参数(durable)：这个队列是否需要持久（即，服务重启后，这个队列是否需要还存在；这儿我们根据自己的需求，设为了true；）
//第三个参数(exclusive)：这个队列是否独有（即，这个队列是不是仅能给这个连接使用；这儿我们设为了false）
//第四个参数(autoDelete)：这个队列是否需要自动删除（即，在队列没有使用的时候，是否需要自动删除；这儿我们设为了false）
//第五个参数(arguments)；
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
System.out.println("开始接收消息");
//这句话的意思，这个消费者最希望处理的消息的数量；那么效果就是，这个消费者在处理完一个消息之前，是不会接收下一个消息的；
channel.basicQos(1);
//6.接收消息，并消费
//参数说明：第一个参数（queue）队列名；
// 第二个参数（autoAck），是否去自动的确认收到；即，这儿接收到消息之后，是否需要通知消息发送者；
//为了演示公平派遣，我们这儿改成了false，即我么手动却发送处理了完成的消息
//第三个参数（callback），消息收到后的处理
channel.basicConsume(TASK_QUEUE_NAME, false, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("收到了消息" + message);
try{
doWork(message);
}finally {
System.out.println("完成消息处理");
//消息处理完成后，去手动确认；
//第一个参数(deliveryTag)这个参数是固定的；
//第二参数(multiple)意思是，我们是否同时多个消息一起确认，这儿我们不需要，所以设为了false；
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
});
}
/**
* 工具方法：处理消息；
* @param task
*/
private static void doWork(String task) {
//根据具体消息内容的不同，去处理消息；
// 即，如果消息中有'.'的话，那么我们就让其处理速度慢1秒；（PS：这儿仅仅是为了演示用的，玩具式程序）
// 那么，这样一来，就会出现这个效果：如果消息中没有'.'，处理的就会很快；如果有'.'，处理速度就会慢的多；
char[] chars = task.toCharArray();
for (char ch : chars) {
if (ch == '.') {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
说明：
（1） 修改内容说明；
（2） 注意：此时，一定要记得主动发送【消息处理完成的通知】；否则，RabbitMQ就不知道，这个消息是否被处理完了，其就会认为没有被处理完，于是后续的消息就会得不到处理，越积越多，消耗内存；
（3） 运行效果；此时，我们重开两个Worker，然后运行NewTask，发送10条消息；
