MQ组件之RabbitMQ入门
同步调用和异步调用
在微服务架构中,服务之间的调用有同步调用和异步调用两种方式。
我们使用OpenFeign去调用是同步调用,同步调用的缺点很明显,在下图的场景中,支付完成后需要调用订单服务、仓库服务、短信服务。没每调用一个服务耗时150ms,那么全部调用完成就需要450ms,如果后面还需要调用更多的服务,那么耗时会更久,而且如果后期需求变更,支付完成后还需要调用其它服务,那么还需要改代码。综上所述
- 同步调用的优点:
- 时效性较强,可以立即得到结果
- 同步调用的缺点:
- 耦合度高
- 性能和吞吐能力下降
- 有额外的资源消耗
- 有级联失败问题
异步调用常见的实现是基于事件驱动的方式。
- 异步调用的优点:
- 耦合度低
- 吞吐量提升
- 故障隔离
- 流量削峰
- 异步调用的缺点:
- 依赖于Broker的可靠性、安全性、吞吐能力
- 架构复杂了,业务没有明显的流程线,不好追踪管理
什么是MQ
MQ (MessageQueue),中文是消息队列,字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。
下图是常见MQ组件的对比。
RabbitMQ安装
这里使用的是docker镜像安装的方式。
- 首先把RabbitMQ镜像上传到虚拟机中,然后加载RabbitMQ镜像
docker load -i /usr/local/mq.tar
- 启动RabbitMQ容器
docker run -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=root -v mq-plugins:/plugins --name mq --hostname mq -p 15672:15672 -p 5672:5672 -d rabbitmq:3-management
输入虚拟机的ip和开放的端口,即可访问RabbitMQ的管理后台:
RabbitMQ管理后台介绍
下图是RabbitMQ管理后台的预览页面,展示了RabbitMQ节点的相关信息,如节点的名字、状态、运行时间等
下图Connections(连接)页面,在这个页面查看、管理和关闭当前所有的TCP连接。
下图是Channels(通道)页面,这里展示了所有当前打开的通道以及它们的详细信息
下图是Exchanges(交换机)页面,这里可以查看、创建和删除交换机。交换机是RabbitMQ消息路由的重要组成部分,它决定了消息如何被路由到正确的队列。
下图是Queues(队列)页面,可以在这里看到所有的队列。队列是RabbitMQ的核心,它保存了所有待处理的消息。
下图是Admin(管理)页面,在这里可以管理RabbimtMQ的用户,并分配相关功能。为了给不同的用户授权不同的功能,RabbitMQ设计了一个virtual host(虚拟主机)的概念,每个用户只能看到它所在虚拟主机下的资源。virtual host是对queue、exchange等资源的逻辑分组。
下面是RabbitMQ官方关于如何监控RabbitMQ以及RabbitMQ管理后台的介绍。
如何监控RabbitMQ
RabbitMQ的管理插件介绍
RabbitMQ相关概念
下图是AMQP的通讯模型,由消息发送者、消息接收者、交换机、队列组成。
- 消息发送者:发送消息
- 消息接收者:消费消息
- 交换机:路由消息,决定消息路由到哪个队列
- 队列:负责存储消息
下图是RabbitMQ对于AMQP通讯模型的实现。
RabbitMQ中常见的通讯模型主要可以分为两种,队列和发布/订阅,而这两种又可以往下细分。
- 队列
- 基本消息队列
- 工作消息队列
- 发布订阅,根据交换机类型不同可以分为三种
- Fanout Exchange:广播
- Direct Exchange:路由
- Topic Exchange:主题
使用
下面介绍RabbitMQ中简单队列的使用。
消息发布者代码
public class Send {private static final String QUEUE_NAME = "hello";public static void send() {//建立和RabbitMQ Server的连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.200.215");connectionFactory.setUsername("root");connectionFactory.setPassword("root");try (Connection connection = connectionFactory.newConnection()) {//创建通道Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);String message = "Hello World1";//发送消息channel.basicPublish("", QUEUE_NAME, null, message.getBytes());} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}public static void main(String[] args) {send();}}
消息接收者代码
public class Recv {private static final String QUEUE_NAME = "hello";public static void receive() {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.200.215");factory.setUsername("root");factory.setPassword("root");try(Connection connection = factory.newConnection()) {Channel channel = connection.createChannel();//这里声明队列是因为,消费者有可能先启动,此时要确保有队列,才能尝试从队列中获取消息channel.queueDeclare(QUEUE_NAME, false, false, false, null);DeliverCallback callback = new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {String message = new String(delivery.getBody(), "UTF-8");System.out.println("receive message: " + message);}};channel.basicConsume(QUEUE_NAME, true, callback, s -> {});} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}public static void main(String[] args) {receive();}}
消息发送成功后,可以在RabbitMQ管理后台中看到
通过下面的Get Message按钮可以获取未消费的消息
参考
- RabbitMQ官方教程
- rabbitMQ入门指南:管理页面全面指南及实战操作