MQ概述:
Message Queue(消息队列),实在消息的传输过程中保存消息的容器,都用于分布式系统之间进行通信
分布式系统通信的两种方式:直接远程调用 和 借助第三昂 完成间接通信
发送方称谓生产者,接收方称为消费者
MQ优势和劣势
优势:
- 应用解耦:如 把一个系统的不同功能拆分出来,中间连接mq,那么其中一个功能的系统挂掉也不会影响其它系统,达到解耦
- 异步提速:如 一个订单接受下单请求,正常需要去库存,支付,物流三个系统走一遍,但是可以通过直接交手mq然后直接返回,那三个系统用mq去排队进行操作,大大缩短等待时间,增加系统吞吐量
- 削峰填谷
劣势:
- 系统可用性降低
- 系统复杂性提高
- 一致性问题
RabbitMQ时基于AMQP(高级信息队里额协议)用于软件通信的中间件
RabbitMQ四大核心
- 生产者
- 消费者
- 队列
- 交换机
AMQP协议是一种二进制协议,它定义了一组规则和标准,以确保消息可以在不同的应用程序和平台之间传递和解释,AMQP包含四个核心组件:
- 消息
- 交换机
- 队列
- 绑定
工作原理
Broker:接受和分发消息的应用。RabbitMQ Server 就是 Message Broker
Virtual host:是一个虚拟主机的概念,一个Broker可以有多个Virtual host,每个Virtual都有一套自己的Exchange(交换机)和Queue(队列)
快速上手
- 导入依赖
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.16.0</version>
</dependency>
- 写一个生产者
public class Producer {
public static void main(String[] args) throws Exception {
String queueName = "test_queue_name";
String exchangeName = "test_exchange_name";
//创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//服务地址
factory.setHost("xx.xx.xx.xx");
//账号密码
factory.setUsername("root");
factory.setPassword("xxx");
//端口号
factory.setPort(5672);
//创建链接
Connection connection = factory.newConnection();
//创建信道
Channel channel = connection.createChannel();
/**
* 创建交换机
* params:
* 1. 交换机名称
* 2. 交换机类型:direct,topic,fanout,headers
* 3. 指定交换机是否持久化 - true:那么交换机的元数据要持久化
* 4. 指定交换机没有队列绑定时是否需要删除
* 5. Map<String,Object>用来指定我们交换机其他的结构化参数,
*/
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);
/**
* 生成一个队列
* params
* 1. 队列名称
* 2. 队列是否需要持久化 - 是队列名称等这些元数据的持久化,不是队列中消息的持久化
* 3. 标识队列是不是私有的,如果是私有的,只有创建它的应用程序才能消费消息
* 4. 队列在没有消费者订阅的情况下是否自动删除
* 5. 队列的一些结构化信息,比如声明私信队列,磁盘队列会用到
*/
channel.queueDeclare(queueName,true,false,false,null);
/**
* 将交换机和队列绑定
* params:
* 1.队列名称
* 2. 交换机名称
* 3. 路由键 - 在直连模式下,可以为我们队列名称
*/
channel.queueBind(queueName,exchangeName,queueName);
/**发送消息
* params:
* 1. 发送到哪个交换机
* 2. 队列名称
* 3. 其他参数信息
* 4. 发送消息的消息体
*/
String msg = "hello RabbitMQ!";
channel.basicPublish(exchangeName,queueName,null,msg.getBytes());
channel.close();
connection.close();
}
}
- 再写一个消费者来消费消息
public class Consumer {
public static void main(String[] args) throws Exception{
String queueName = "test_queue_name";
String exchangeName = "test_exchange_name";
//创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//服务地址
factory.setHost("xx.xx.xx.xx");
//账号密码
factory.setUsername("root");
factory.setPassword("xxx");
//端口号
factory.setPort(5672);
//创建链接
Connection connection = factory.newConnection();
//创建信道
Channel channel = connection.createChannel();
//接收消息的回调函数
DeliverCallback deliverCallback = (consumerTage, message) -> {
System.out.println("接收到消息" + new String(message.getBody()));
};
//取消消息的回调函数
CancelCallback cancelCallback = consumerTage -> {
System.out.println("消费消息被中断");
};
/**
* 消费消息
* Params:
* 1. 消费那个队列
* 2. 消费成功后是否自动应答
* 3. 接收消息的回调函数
* 4. 取消消息的回调函数
*/
channel.basicConsume(queueName,true,deliverCallback,cancelCallback);
}
}
RabbitMQ的交换机类型
-
direct
路由键与队列名完全匹配的交换机,通过RoutingKey路由键将交换机和队列进行绑定
public class Producer {
public static void main(String[] args) throws Exception {
String exchangeName = "test_exchange_name";
String queueName1 = "test_queue_name1";
String queueName2 = "test_queue_name2";
String queueName3 = "test_queue_name3";
String queueName4 = "test_queue_name4";
String key1 = "key_1";
String key3 = "key_3";
String key4 = "key_4";
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("xx.xxx.xxx.xx");
factory.setUsername("root");
factory.setPassword("xxx");
factory.setPort(5672);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);
channel.queueDeclare(queueName1,true,false,false,null);
channel.queueDeclare(queueName2,true,false,false,null);
channel.queueDeclare(queueName3,true,false,false,null);
channel.queueDeclare(queueName4,true,false,false,null);
channel.queueBind(queueName1,exchangeName,key1);
channel.queueBind(queueName2,exchangeName,key1);
channel.queueBind(queueName3,exchangeName,key3);
channel.queueBind(queueName4,exchangeName,key4);
channel.basicPublish(exchangeName,key1,null,"hello Key1".getBytes());
channel.basicPublish(exchangeName,key3,null,"hello Key3".getBytes());
channel.basicPublish(exchangeName,key4,null,"hello Key4".getBytes());
channel.close();
connection.close();
System.out.println("发送消息成功");
}
}
如上述代码,执行过后,四个队列每个都会收到消息,其中test_queue_name1,test_queue_name2各收到了一条hello Key1
direct的拼配要求是精准的,设置路由为key1就只能向key1发送才会接收到,发送key或者key.1之类都是不会接收到的
2. fanout
扇出类型交换机,会将消息分发给所有绑定了此交换机的队列,此时RoutingKey参数无效
也就是说,fanout类型交换机下发一条消息,无论RoutingKey是什么,所有队列都会收到消息
-
topic
主题类型交换机,此种交换机与Direct类似,也需要routingkey进行匹配分发,区别在于可以模糊查询
Topic中,将routingkey通过”.”来分为多个部分
“*”:代表一个部分
“#”:代表0-多个部分,路由键为#,则代表接受所有信息 -
headers 性能太差 不做学习