当前位置：首页 > news >正文

[RabbitMQ] RabbitMQ简单概述，用法和交换机模型

news 2025/8/16 12:49:24

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(队列)

RabbitMQ工作原理

快速上手

导入依赖

    <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("发送消息成功");}
}