大家好我是苏麟今天带来rabbitmq.
RabbitMQ官网 : RabbitMQ: easy to use, flexible messaging and streaming — RabbitMQ
同步和异步通讯
微服务间通讯有同步和异步两种方式:
同步通讯:就像打电话,需要实时响应。
异步通讯:就像发邮件,不需要马上回复。
两种方式各有优劣,打电话可以立即得到响应,但是你却不能跟多个人同时通话。发送邮件可以同时与 多个人收发邮件,但是往往响应会有延迟。
我们之前学习的Feign调用就属于同步方式,虽然调用可以实时得到结果,但存在下面的问题
总结:
同步调用的优点:
同步调用的问题:
异步调用则可以避免上述问题:
我们以购买商品为例,用户支付后需要调用订单服务完成订单状态修改,调用物流服务,从仓库分配响 应的库存并准备发货。
在事件模式中,支付服务是事件发布者(publisher),在支付完成后只需要发布一个支付成功的事件 (event),事件中带上订单id。 订单服务和物流服务是事件订阅者(Consumer),订阅支付成功的事件,监听到事件后完成自己业务 即可。
为了解除事件发布者与订阅者之间的耦合,两者并不是直接通信,而是有一个中间人(Broker)。发布 者发布事件到Broker,不关心谁来订阅事件。订阅者从Broker订阅事件,不关心谁发来的消息。
Broker 是一个像数据总线一样的东西,所有的服务要接收数据和发送数据都发到这个总线上,这个总线 就像协议一样,让服务间的通讯变得标准和可控。
好处:
缺点:
好在现在开源软件或云平台上 Broker 的软件是非常成熟的,比较常见的一种就是我们今天要学习的MQ 技术。
MQ,中文是消息队列(MessageQueue),字面来看就是存放消息的队列。也就是事件驱动架构中的 Broker。
比较常见的MQ实现:
几种常见MQ的对比:
追求可用性:Kafka、 RocketMQ 、RabbitMQ
追求可靠性:RabbitMQ、RocketMQ
追求吞吐能力:RocketMQ、Kafka
追求消息低延迟:RabbitMQ、Kafka
MQ的基本结构:
RabbitMQ中的一些角色:
RabbitMQ官方提供了5个不同的Demo示例,对应了不同的消息模型:
结构如下:
包括三部分:
引入依赖
-
-
com.github.luues -
spring-boot-starter-rabbitmq -
1.3.0.5.RELEASE -
简单队列模式的模型图:
官方的HelloWorld是基于最基础的消息队列模型来实现的,只包括三个角色:
思路:
- package com.sl;
-
- 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 {
- @Test
- public void testSendMessage() throws IOException, TimeoutException {
- // 1.建立连接
- ConnectionFactory factory = new ConnectionFactory();
- // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
- factory.setHost("192.168.75.128");
- factory.setPort(5672);
- factory.setVirtualHost("/");
- factory.setUsername("root");
- factory.setPassword("root");
- // 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.发送消息
- String message = "hello, rabbitmq!";
- channel.basicPublish("", queueName, null, message.getBytes());
- System.out.println("发送消息成功:【" + message + "】");
- // 5.关闭通道和连接
- channel.close();
- connection.close();
- }
- }
代码思路:
- package com.sl;
-
- 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.75.128");
- factory.setPort(5672);
- factory.setVirtualHost("/");
- factory.setUsername("root");
- factory.setPassword("root");
- // 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("等待接收消息。。。。");
- }
- }
总结
基本消息队列的消息发送流程:
基本消息队列的消息接收流程:
下阶段 : Spring AMQP-CSDN博客
这期就到这里下期见 !