【SpringCloud】什么是MQ

RabbitMQ

1.什么是MQ

1.1 同步和异步通信

微服务间通讯有同步和异步两种方式：

• 同步通讯：就例如打电话一样，需要实时响应
• 异步通讯：就例如发邮件一样，不需要马上回复

1.1.1同步通讯

Feign的远程调用就属于同步通讯，虽然可以得到实时响应，但是也存在一些问题

同步调用的优点：时效性较强，可以立即得到结果

同步调用的问题：

• 耦合度高
• 性能和吞吐能力下降
• 有额外的资源消耗
• 有级联失败问题

1.1.2异步通讯

异步调用可以避免同步调用出现的问题，发布方和接收方两者不是直接通信，而是有一个中间件，它就像数据总线一样，所有的服务要接收数据和发送数据都发到这个总线上，它让服务见的通讯变得标准和可控。

优点：

• 吞吐量提升：无需等待订阅者处理完成，响应更快速
• 故障隔离：服务没有直接调用，不存在级联失败问题
• 调用间没有阻塞：不会造成无效的资源占用
• 耦合度极低：每个服务都可以灵活插拔，可替换
• 流量削峰：不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件

缺点：

• 架构复杂化，业务没有明显的流程先，不好管理
• 需要依赖于Broker的可靠、安全、性能

1.2 技术对比

MQ，中文是消息队列（MessageQueue），字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

比较常见的MQ实现：

• ActiveMQ
• RabbitMQ
• RocketMQ
• Kafka

几种常见MQ的对比：

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

2.快速入门

MQ基本结构：

RabbitMQ中的一些角色：

• publisher：生产者
• consumer：消费者
• exchange：交换机，负责消息路由
• queue：队列，存储消息
• virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离

2.1RabbitMQ消息模型

RabbitMQ官方提供了5个不同的示例，对应了不同的消息模型：

• 基本消息队列（BasicQueue）
• 工作消息队列（WorkQueue）
• 发布订阅，又根据交换机类型不同分为三种：
  • Fanout Exchange：广播
  • Direct Exchange：路由
  • Topic Exchange：主题

2.4.1.publisher实现

简单队列模式的模型图：

步骤：

• 建立连接
• 创建Channel
• 声明队列
• 发送消息
• 关闭连接和channel

代码实现：

package com.example.mq.helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class PublisherTest {@Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.150.101");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("haojiale");factory.setPassword("123321");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.发送消息String message = "hello, rabbitmq!";channel.basicPublish("", queueName, null, message.getBytes());System.out.println("发送消息成功：【" + message + "】");// 5.关闭通道和连接channel.close();connection.close();}
}

2.4.2.consumer实现

步骤：

• 建立连接
• 创建Channel
• 声明队列
• 订阅消息

代码实现：

package com.example.mq.helloworld;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.150.101");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("haojiale");factory.setPassword("123321");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){@Overridepublic 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("等待接收消息。。。。");}
}