今天学习RabbitMQ，你知道RabbitMQ是什么吗，RabbitMQ是一种消息中间件，我们在写很多业务的时候，有时候我们需要考虑到消息的实时性，时效性和一致性，比如说我们最熟悉的订单业务，一个订单从下单到订单到库存发货都是有实时性的，如果中间取消订单我们还要保持一致性，不能前面下订单了后面取消了给人发货了吧，但是如果用普通的java代码来做，那肯定很麻烦，解决麻烦的途径就是找一个好用的框架，那今天我们就一起来学习下RabbitMQ这个框架。
这里我看的教程是因为Rabbitmq是基于Erlang环境，先安装Erlang然后再安装Rabbitmq就很麻烦，说是docker比较方便，那我们就用docker嘛
这里还不会docker的可以看下，一文教会你docker：https://blog.csdn.net/hello_list/article/details/124221409
如果不会linux，一文教会：https://blog.csdn.net/hello_list/article/details/123977208
很简单只需要一行代码，搞定
docker run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq -v /mydata/rabbitmq/data:/mydata/rabbitmq/data rabbitmq
容器创建完毕，同时也在本地跑起来了；我们可以去访问web界面
访问失败
应该是我这个防火墙端口号没有打开，把防火墙直接关了
systemctl stop firewalld
systemctl disable firewalld
还是访问不了，然后我去查了下好像这个不带web界面，到了官方这里，官方地址：https://www.rabbitmq.com/download.html
换一下，换了一个镜像，把之前的删除了重新创建容器
docker run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=root -v /mydata/rabbitmq/data:/mydata/rabbitmq/data rabbitmq:3.10-management
访问测试成功
rootroot登录，访问成功
接下来我们学习，就可以通过这个web界面进行操作，包括查看与代码结合使用；
安装好之后我们就开始学习
官网：https://www.rabbitmq.com/，好多兔子
来到GetStarted我们看到Rabbitmq支持7中模式，比如我们要学习什么，java我们点进去学习就可以，你学习别的语言python都可以直接去官方看着学就可以，很方便，而且一定错不了，还是最新的；
我们点进去java看下简单模式，看官网说明看的懂哈，一个producter然后发送消息进入消息队列，然后被consumer进行消费；
“P” is our producer and “C” is our consumer.
话不多说，代码好懂，我们写一个简单例子出来
环境说明：jdk1.8，idea2019
1、创建一个普通的maven工程
这里就不用说怎么创建了吧
2、导入依赖，java原生依赖，一定要检查依赖导入成功
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.10.0</version>
</dependency>
3、创建一个生产者提供服务，这里是给自己深化理解的，参考：https://www.kuangstudy.com/zl/rabbitmq#1366709584634437634
public class Producer {
public static void main(String[] args) {
// 1: 创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
// 2: 设置连接属性
connectionFactory.setHost("192.168.43.12");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("root");
connectionFactory.setPassword("root");
Connection connection = null;
Channel channel = null;
try {
// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("生产者");
// 4: 从连接中获取通道channel
channel = connection.createChannel();
// 5: 申明队列queue存储消息
/*
* 如果队列不存在，则会创建
* Rabbitmq不允许创建两个相同的队列名称，否则会报错。
*
* @params1： queue 队列的名称
* @params2： durable 队列是否持久化
* @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
* @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
* @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
* */
channel.queueDeclare("queue1", false, false, false, null);
// 6： 准备发送消息的内容
String message = "你好，学相伴！！！";
// 7: 发送消息给中间件rabbitmq-server
// @params1: 交换机exchange
// @params2: 队列名称/routing
// @params3: 属性配置
// @params4: 发送消息的内容
channel.basicPublish("", "queue1", null, message.getBytes());
System.out.println("消息发送成功!");
} catch (Exception ex) {
ex.printStackTrace();
System.out.println("发送消息出现异常...");
} finally {
// 7: 释放连接关闭通道
if (channel != null && channel.isOpen()) {
try {
channel.close();
} catch (Exception ex) {
ex.printStackTrace();
}
}
if (connection != null) {
try {
connection.close();
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
}
}
4、创建一个消费者消费服务
public class Consumer {
public static void main(String[] args) {
// 1: 创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
// 2: 设置连接属性
connectionFactory.setHost("192.168.43.12");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("root");
connectionFactory.setPassword("root");
//获取队列的名称
final String queueName = Thread.currentThread().getName();
Connection connection = null;
Channel channel = null;
try {
// 3: 从连接工厂中获取连接
connection = connectionFactory.newConnection("生产者");
// 4: 从连接中获取通道channel
channel = connection.createChannel();
// 5: 申明队列queue存储消息
/*
* 如果队列不存在，则会创建
* Rabbitmq不允许创建两个相同的队列名称，否则会报错。
*
* @params1： queue 队列的名称
* @params2： durable 队列是否持久化
* @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
* @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
* @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
* */
// 这里如果queue已经被创建过一次了，可以不需要定义
//channel.queueDeclare("queue1", false, false, false, null);
// 6： 定义接受消息的回调
Channel finalChannel = channel;
finalChannel.basicConsume("queue1", true, new DeliverCallback() {
@Override
public void handle(String s, Delivery delivery) throws IOException {
System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
}
}, new CancelCallback() {
@Override
public void handle(String s) throws IOException {
}
});
System.out.println(queueName + "：开始接受消息");
System.in.read();
} catch (Exception ex) {
ex.printStackTrace();
System.out.println("接收消息出现异常...");
} finally {
// 7: 释放连接关闭通道
if (channel != null && channel.isOpen()) {
try {
channel.close();
} catch (Exception ex) {
ex.printStackTrace();
}
}
if (connection != null && connection.isOpen()) {
try {
connection.close();
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
}
}
测试
运行生产者生产消息，消息发送成功：
我们在图形化界面上直接看到有一条消息产生：
web界面可以接收消息，这里大家看，这里有一个nack和一个ack，nack可以预览小，模拟接收消息，但是不会真的消费了这个消息，而ack就是消费消息，消费了就是没有了，这里大家可以用nack预览消息，但是切记不要手动消费消息，不然消息就没有了
我们可以使用nack预览消息，成功接收消息，同时消息是还在队列里面：
我们还可以看看web图形化界面如何操作，大家在使用的时候可以结合web图形化界面使用，很方便；
接下来我们使用程序来接收消息：
大家注意一个细节啊，生产完发送消息，程序就会停止了，但是接收消息，我们接收到消息，程序并没有停止，所以当我们对一个队列进行消息接受的时候，除非程序停止，不然一直会消费和接收队列中的消息，而生产者只负责生产发送出去消息就会立即停止；
官网文档：https://www.rabbitmq.com/tutorials/tutorial-two-java.html
刚才我们学习了点对点，生产者消费者在一个队列中生成发布消息和消费消息，那多个消费者，我们也是有两个机制的
轮询分发（默认）
公平分发
消费者修改下代码即可，改成手动应答：
Manual message acknowledgments are turned on by default. In previous examples we explicitly turned them off via the autoAck=true flag. It's time to set this flag to false and send a proper acknowledgment from the worker, once we're done with a task.
channel.basicQos(1); // accept only one unack-ed message at a time (see below)
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + message + "'");
try {
doWork(message);
} finally {
System.out.println(" [x] Done");
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
};
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });
我们接下来学习下交换机的四种模式，交换机具有消息发布的功能
大家不要看这几个模式什么的，不要太复杂，等看完你就知道差别不大，我们通过这张图可以看一下，他其实就是加了一个路由，我们可以指定路由进行消息的发送到不同的队列，由不用的消费者消费；
程序代码也不会有太大变化啊
生产者
我们通过web界面去创建一个direct模式的交换机，交换机，就是上图中的那个x我们只需要给指定交换机中发送消息，然后交换机通过路由分发到不同的队列，不用想想太复杂，一听名字交换机就认为很难一样，就想刚才我们使用simple就没有用到交换机，不是没有用到，是我们直接用的默认交换机，default就是；
然后我们创建几个队列，分别指定不用的路由key，也是直接在命令行这里创建吧
添加队列
点进创建的队列，然后我们可以给这个队列绑定上我们创建的交换机
这样我们就创建了三个队列，然后都绑定到了这个交换机，相信大家一看就知道这是干什么，这是当我们有个消息过来，我们要发给微信还是qq还是邮件
现在我们给交换机推送一条信息，那这条消息呢我们就通知到微信
这样我们就看到微信的队列就多出了一条消息
我们可以预览下这条消息，就是我们发的
这里我就不上代码了吧，其实这些都很简单的，官网上有代码，看官网上代码吧，其实就跟之前的代码一样呀，就是创建了一个交换机和队列指定了下路由key，主要就是加上这几个代码吧，官网：https://www.rabbitmq.com/tutorials/tutorial-three-java.html
channel.exchangeDeclare("logs", "fanout"); // 创建一个交换机，param1 交换机name,param2交换机模式
String queueName = channel.queueDeclare().getQueue(); // 创建队列
channel.queueBind(queueName, "logs", ""); // 交换机与队列绑定
上面创建交换机或者队列，如果存在再创建就回报错的，就这些大家自己去尝试好了；
其实都是一次性的，比如创建和绑定我们是可以通过web界面这样操作的，很方便，不然你写这个创建的代码也是一次性的嘛！
其实就是在direct模式上添加了模糊匹配
.* // 表示一个单词
.# // 表示没有单词或者一个或多个单词
在交换机是direct模式中呢，也就是刚才我们发现，想要跟微信，qq队列推送消息，相当于要个交换机push两条消息，指定路由key分别为微信和qq，是不是感觉有点儿不足呢，我们不能达到一次推送多个分发的效果呀，这个时候我们就可以用Topics模式，带着这个任务，我们去修改下，使用Topics模式；
首先肯定是先创建交换机了
我们重新绑定刚才的三个队列，指定上路由key，这里我就不在创建三个队列了，我们还是拿这三个队列try下
这样我们就绑定玩关系了，那#和*分别是什么意思，刚开始就有说，那我为什么这么写呢，我这里有一个需求，邮件不能单独发，也就是你可以单独发qq或者单独发微信，但是你如果要发邮件的话，必须是发了qq或者微信，就这样。
我们试下，比如我们给qq和微信发消息
可以看到qq和微信都收到了消息
你也可以尝试下单独给qq发，单独给微信发，这里我尝试下单独给email发，看到了吧，失败了
那我们要想发邮件就必须带上qq或者微信，尝试下
需求实现，完成
代码呢，还是自己去看下吧，都一样，大差不差几行代码，官方文档：https://www.rabbitmq.com/tutorials/tutorial-five-java.html
其实基本上所有课程都是首先讲fanout模式，然后层层带入，fanout模式最简单，创建出一个fanout交换机，绑定上几个队列，交换机发送消息的时候，绑定的多个队列就都会收到消息，是不是很简单，我们就是要把难点儿的放在前面讲嘛，这样不就很简单了；
headers模式几乎不怎么用，其实就是，看到下面的参数了嘛，交换机发送消息的时候，参数对应上就可以，没得讲，而且我听说几乎不怎么用，所以，基本差不多；
今天先到这里，学习一下基本的消息收发，那后面呢我们就可以简化，跟springboot整合，我们知道什么跟springboot整合之后就回简单许多，那这里的全部大家都要去实现一下，然后我们后面再学习比如如何结合业务逻辑，结合订单模块等等，那看到这里了好不点个赞，点个关注嘛，bye~