RabbitMQ
一、MQ
1.同步调用的优缺点
同步调用的优点：
- 时效性较强，可以立即得到结果
同步调用的问题：
- 耦合度高
- 性能和吞吐能力下降
- 有额外的资源消耗
- 有级联失败问题
2.异步调用
异步调用常见实现就是事件驱动模式
好处：
-
吞吐量提升：无需等待订阅者处理完成，响应更快速
-
故障隔离：服务没有直接调用，不存在级联失败问题
-
调用间没有阻塞，不会造成无效的资源占用
-
耦合度极低，每个服务都可以灵活插拔，可替换
-
流量削峰：不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件
缺点：
- 架构复杂了，业务没有明显的流程线，不好管理
- 需要依赖于Broker的可靠、安全、性能
3.MQ实现
MQ，中文是消息队列（MessageQueue），字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。
几种常见MQ的对比：
| RabbitMQ | ActiveMQ | RocketMQ | Kafka | |
|---|---|---|---|---|
| 公司/社区 | Rabbit | Apache | 阿里 | Apache |
| 开发语言 | Erlang | Java | Java | Scala&Java |
| 协议支持 | AMQP，XMPP，SMTP，STOMP | OpenWire,STOMP，REST,XMPP,AMQP | 自定义协议 | 自定义协议 |
| 可用性 | 高 | 一般 | 高 | 高 |
| 单机吞吐量 | 一般 | 差 | 高 | 非常高 |
| 消息延迟 | 微秒级 | 毫秒级 | 毫秒级 | 毫秒以内 |
| 消息可靠性 | 高 | 一般 | 高 | 一般 |
追求可用性：Kafka、 RocketMQ 、RabbitMQ
追求可靠性：RabbitMQ、RocketMQ
追求吞吐能力：RocketMQ、Kafka
追求消息低延迟：RabbitMQ、Kafka
二、Linux安装RabbitMQ
1.1.下载镜像
方式一：在线拉取
docker pull rabbitmq:3-management
方式二：从本地加载
在课前资料已经提供了镜像包：
上传到虚拟机中后，使用命令加载镜像即可：
docker load -i mq.tar
1.2.安装MQ
执行下面的命令来运行MQ容器：
docker run \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq \
--hostname mq1 \
-p 15672:15672 \
-p 5672:5672 \
-d \
rabbitmq:3-management
注意5672是MQ消息通信的端口，15672是ui端口
三、RabbitMQ
1.种类
大致分为两种
1.**无交换机的 **基本消息队列 和 工作消息队列
2.有交换机的发布订阅，路由，主题
2.基于SpringAMQP使用
2.1基础消息队列
-
先使用测试类生成队列
public class ConsumerTest { public static void main(String[] args) throws IOException, TimeoutException { // 1.建立连接 ConnectionFactory factory = new ConnectionFactory(); // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码 factory.setHost("116.62.32.68"); factory.setPort(5672); factory.setVirtualHost("/"); factory.setUsername("itcast"); factory.setPassword("123321"); // 1.2.建立连接 Connection connection = factory.newConnection(); // 2.创建通道Channel Channel channel = connection.createChannel(); // 3.创建队列 String queueName = "simple.queue"; channel.queueDeclare(queueName, false, false, false, null); // 4.订阅消息 channel.basicConsume(queueName, true, new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // 5.处理消息 String message = new String(body); System.out.println("接收到消息：【" + message + "】"); } }); System.out.println("等待接收消息。。。。"); } } -
引入依赖
<!--AMQP依赖，包含RabbitMQ--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> -
生产者编写yaml文件的MQ地址，端口号，用户名密码
spring: rabbitmq: host: 192.168.150.101 # 主机名 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: itcast # 用户名 password: 123321 # 密码 -
编写生产者代码
@RunWith(SpringRunner.class) @SpringBootTest public class SpringAmqpTest { //类似于redisTemplate @Autowired private RabbitTemplate rabbitTemplate; @Test public void testSimpleQueue(){ //队列名 String queueName = "simple.queue"; //消息 String message = "hello,SpringAmqp"; //发送消息到队列 rabbitTemplate.convertAndSend(queueName,message); } } -
消费者编写yaml文件的MQ地址，端口号，用户名密码
同上
-
编写消费者代码
@Component public class SpringRabbitListener { @RabbitListener(queues = "simple.queue") public void listenSimpleQ