视频就不行了异步通讯不要求实时性就好像你用手机发短信好多人都能同时给你发短信你都可以收到而且不用及时回复。

比如用户调用支付服务时它需要先后调用订单服务、仓储服务、短信服务等都调用结束后支付服务再返回用户相关信息故这个过程的响应时间实际上就是所有这些相关服务执行之后所用时间之和这样是非常影响效率的。

但是也有优点时效性较强可以立即得到结果

1.如果我们想对支付服务增加一些功能增加一些别的服务为了让支付功能调用这个新服务我们需要改动相关的代码

2.调用者需要等待服务提供者响应如果调用链过长则响应时间等于每次调用的时间之和。

就如支付服务必须要等订单服务、仓储服务、短信服务都执行后才能给出响应很慢。

3.上述支付服务在等待其余服务响应的时候资源是不释放的一直在等待高并发场景下嫉妒浪费系统资源

4.如果其中一个服务出现问题例如仓储服务出现问题则整个响应就无法进行了就整个崩掉了。

异步调用方案

比如用户发起支付服务此时会将该消息发给Broker然后就不用管后面了支付服务直接给用户响应不需要管订单、仓储、短信这些服务的执行了这几个服务提前订阅了这个Broker所以支付服务消息来了会从Broker中获取到消息然后执行自己的逻辑就好。

优势一:服务解耦

当我们对支付服务增加一些新的功能新服务时只需要将该服务订阅这个Broker即可不需要改动支付服务的代码

优势二:性能提升吞吐量提高

因为是异步调用支付服务不需要关注其相关的其他服务执行只需将支付消息发给Broker即可然后就能直接给用户响应非常快后续相关的服务只需要从Broker中收到该消息后执行相关的业务操作就可。

优势三:服务没有强依赖不担心级联失败问题

其中一个服务发生故障整个支付服务不会因为这个服务故障了而停掉支付服务依然会直接给用户响应。

优势四:流量削峰

因为有Broker的缘故如果并发量比较大这些消息都会暂存在Broker中慢慢的让其余各服务去处理所以能够很好的将流量削峰。

MQ(MessageQueue)中文是消息队列字面来看就是存放消息的队列。

也就是事件驱动架构中的Broker。

简单来讲Kafka对并发场景性能更高但是没有RabbitMQ安全所以一般情况下就用RabbitMQ就可中规中矩。

RabbitMQ快速入门

这里第一个15672端口是用来登录RabbitMQ控制台的第二个5672端口是用来进行异步调用的。

这里也要设置RabbitMQ控制台的登录账号itcast密码123321

用宿主机ip第一个端口15672来访问RabbitMQ控制台账号itcast密码123321登录即可

exchange:路由消息到队列中

MQ的官方文档中给出了5个MQ的Demo示例对应了几种不同的用法

HelloWorld案例

官方的Helloworld是基于最基础的消息队列模型来实现的只包括三个角色:

queue:消息队列负责接受并缓存消息

编写publisher消息发布者代码新建一个连接工厂配置本机RabbitMQ的异步调用端口以及账号密码然后建立一个连接connection对象。

连接对象connection生成后我们可以在RabbitMQ的控制台看到了这个新生成的连接

然后用这个connection连接对象创建一个通道channel对象

然后用channel对象新建一个名为simple.queue的队列并给这个队列发消息

可以在控制台看到这个新创建的名为simple.queue的队列并且队列中已经有了我们刚发送的消息

订阅该队列的consumer编写前三步与publisher一致

最后一步变成了利用channel将消费者与队列绑定在handleDelivery中定义consumer的消费行为

SpringAMQP

案例利用SpringAMQP实现HelloWorld中的基础消息队列功能

流程如下:

2.在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列

3.在consumer服务中编写消费逻辑绑定simple.queue这个队列

步骤1:

因为publisher和consumer服务都需要amqp依赖因此这里把依赖直接放到父工程mq-demo中

步骤2:

在publisher中编写测试方法向simple.queue发送消息

1.在publisher服务中编写application.yml添加mq连接信息

SpringAMOP如何发送消息?

利用RabbitTemplate的convertAndSend方法

步骤3:

1.在consumer服务中编写application.yml添加mq连接信息

SpringAMQP如何接收消息?

类中声明方法添加RabbitListener注解方法参数就是消息

注意:消息一旦消费就会从队列删除RabbitMQ没有消息回溯功能

Work

queue工作队列可以提高消息处理速度避免队列消息堆积也就是两个消费者同时订阅一个队列共同处理队列中的消息

基本思路如下

1.在publisher服务中定义测试方法每秒产生50条消息发送到simple.queue

2.在consumer服务中定义两个消息监听者都监听simple.queue队列

publisher服务中定义测试方法每20ms发送一个消息发送50次也就是1秒向消息队列中发送50条消息

定义两个消息监听者都监听simple.queue队列消费者1一秒钟能处理50条消息消费者2一秒钟能处理5条消息故理论上这两个消费者能在1秒内处理完publisher发送的所有50条)消息

但是运行起来发现并不是按照能力强的处理的消息多这样来分配的而是这两个监听者各分到了一般的消息消费者1处理偶数号的消息消费者2处理奇数的消息这样最终结果就是消费者1很快的处理完了总共消息的一半25条消息而消费者2却花了好多秒去处理分给自己25条消息最终的结果就是这两个消费者不能够在1秒内处理完所有消息。

实际上它是有一个预取机制造成的这样结果进入到消息队列中的消息会被提前预取给消费者默认的是一人一个这样平均分配的消费者还没处理完但是消息都已经全部预取出来给了对应消费者我们可以配置预取机制让每个消费者预取消息的数量为1每次只能获取一消息处理完成才能获取下一个消息这样就可以做到能者多劳的结果。

消费预取限制

修改application.yml文件设置preFetch这个值可以控制预取消息的上限

配置完后再启动就发现了消费者1执行的消息要比消费者2执行的消息多了而不是简单的将消息都平均分配给监听的两个消费者。

Publish

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。

实现方式是加入了exchange(交换机)。

常见exchange类型包括:

Exchange会将接收到的消息路由到每一个跟其绑定的queue

案例利用SpringAMQP演示FanoutExchange的使用

实现思路如下:

1.在consumer服务中利用代码声明队列、交换机并将两者绑定

2.在consumer服务中编写两个消费者方法分别监听fanout.queue1和fanout.queue2

3.在publisher中编写测试方法向itcast.fanout发送消息

步骤1:

在consumer服务常见一个类添加Configuration注解并声明FanoutExchange、Queue和绑定关系对象Binding代码如下

启动后可以在RabbitMQ控制台看到我们声明的交换机与两个队列完成绑定。

步骤2:

在consumer服务的SpringRabbitListener类中添加两个方法分别监听fanout.queue1和fanout.queue2

步骤3:

在publisher服务发送消息到FanoutExchange

在publisher服务的SpringAmqpTest类中添加测试方法

发布订阅-DirectExchange

会将接收到的消息根据规则路由到指定的Queue因此称为路由模式

(routes)。

每一个Queue都与Exchange设置一个BindingKey

Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

案例利用SpringAMQP演示DirectExchange的使用

实现思路如下:

1.利用RabbitListener声明Exchange、Queue、RoutingKey

2.在consumer服务中编写两个消费者方法分别监听direct.queue1和direct.queue2

3.在publisher中编写测试方法向itcast.direct发送消息

步骤1:

1.在consumer服务中编写两个消费者方法分别监听direct.queue1和direct.queue2.

2.并利用RabbitListener声明Exchange、Queue、RoutingKey

步骤2:

在publisher服务发送消息到DirectExchange

在publisher服务的SpringAmqpTest类中添加测试方法convertAndSend的第二个参数就是RoutingKey如果是red,则两个消费者都可收到是blue,则只有消费者1收到yellow则只有消费者2收到。

如果多个队列具有相同的RoutingKey则与Fanout功能类似

基于RabbitListener注解声明队列和交换机有哪些常见注解?

Queue

TopicExchange与DirectExchange类似区别在于routingKey必须是多个单词的列表并且以.分割。

Queue与Exchange指定BindingKey时可以使用通配符:

#:代指0个或多个单词

案例利用SpringAMQP演示TopicExchange的使用

实现思路如下:

1.并利用RabbitListener声明Exchange、Queue、RoutingKey

2.在consumer服务中编写两个消费者方法分别监听topic.queue1和topic.queue2

3.在publisher中编写测试方法向itcast.topic发送消息

步骤1:

1.在consumer服务中编写两个消费者方法分别监听topic.queue1和topic.queue2

2.并利用RabbitListener声明Exchange、Queue、RoutingKey

步骤2:

在publisher服务的SpringAmqpTest类中添加测试方法如果RoutingKey是china.news则两个消费者都能够接收到消息如果为china.weather,则只有消费者1能收到消息如果是usa.news则只有消费者2能收到消息

消息转换器

说明:在SpringAMQP的发送方法中接收消息的类型是object也就是说我们可以发送任意对象类型的消息SpringAMOP会帮我们序列化为字节后发送

先在consumer中的fanoutConfig中声明一个队列object.queue

执行完publisher发消息后RabbitMQ控制台查看object.queue队列中刚收到的消息看到为乱码被序列化了

Spring的对消息对象的处理是由org.springframeworkamqp.support.converterMessageConverter来处理的。

而默认实现是simpleMessageConverter基于]DK的ObjectOutputStream完成序列化。

我们在publisher服务引入依赖

我们在publisher服务声明MessageConverter在该服务的主启动函数中声明即可。

再次重新执行完publisher发消息后RabbitMQ控制台查看object.queue队列中刚收到的消息看到消息已经成为了json格式而不是乱码了

我们在consumer服务定义MessageConverter在该服务的主启动函数中声明即可

利用MessageConverter实现的默认是JDK的序列化

注意发送方与接收方必须使用相同的MessageConverter