一同步通信

1.同步通信的问题

微服务间基于Feign的调用就属于同步方式，存在一些问题

耦合度高：每次加入新的需求，都需要修改原来党的代码
性能下降：调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和
资源浪费：调用链中的每个服务在等待响应的过程中，不能释放请求占用的资源，高并发场景下会极度浪费资源
级联失败：如果服务提供者出现问题，所有调用方法都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障

同步调用的问题

2.同步调用的优点

时效性强，可以立即得到结果

二异步调用

异步调用常见实现就是事件驱动模式

异步调用的方案

1.异步通信的优点

服务解耦
性能提升，吞吐量提高
服务没有强依赖，不担心级联失败问题
流量削峰

2.异步通信的缺点

依赖于Broker的可靠性、安全性、吞吐能力
架构复杂了，业务没有明显的流程线，不好追踪管理

三 RabbitMQ 快速入门

1.什么是MQ？

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

属性	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP,XMPP.SMTP,STOMP	OpenWire,STOMP,REST,XMPP,AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

2.RabbitMQ 概述

RabbitMQ是基于Erlang语言开发的消息通信中间件，RabbitMQ官网地址：https://www.rabbitmq.com。

3.单机部署RabbitMQ

# 拉取RabbitMQ镜像
docker pull rabbitmq:3-management
# 运行MQ容器
docker run \
  -e RABBITMQ_DEFAULT_USER=itcast \
  -e RABBITMQ_DEFAULT_PASS=123321 \
  --name mq \
  --hostname mql \
  -p 15672:15672 \
  -p 5672:5672 \
  -d \
  rabbitmq:3-management

4.RabbitMQ概述

RabbitMQ概念和结构
RabbitMQ概念和结构

5.RabbitMQ中的几个概念

channel：操作MQ的工具
exchange：路由消息到队列中
queue：缓存消息
virtual host：虚拟主机，是对queue、exchange等资源的逻辑分组

6.常见消息模型（参考链接)

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

(1) HelloWorld案例

HelloWorld是基于组基础的消息队列模型来实现的，只包括三个角色：

publisher：消息发布者，将消息发送到队列queue
queue：消息队列，负责接受并缓存消息
consumer：订阅队列，处理队列中的消息

基本消息队列的消息发送流程：

1.建立connection
2.创建channel
3.利用channel声明队列
4.利用channel向队列发送消息

基本消息队列的消息接收流程：

1.建立connection
2.创建channel
3.利用channel声明队列
4.定义consumer的消费行为handleDelivery()
5.利用channel将消费者与队列绑定

四 SpringAMQP

1.什么是SpringAMQP？

AMQP(Advance Message Queuing Protocol) 是用于在应用程序之间传递业务
消息的开发标准。该协议与语言和平台无关，更符合微服务中独立性的要求

SpringAMQP 是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。
包含两部分，其中spring、-amqp是基础抽象。spring-rabbit是底层默认实现。

2.Basic Queue工作队列

基本思路：

1.在父工程中引入spring-amqp的依赖
2.在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列
3.在consumer服务中编写消费逻辑，绑定simple.queue这个队列

步骤：

1.引入AMQP依赖

<!--        AMQP依赖，包含RabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

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

# publisher的application.yml
spring:
  rabbitmq:
    host: 192.168.127.131 #rabbitMQ的IP地址
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: /

// 单元测试
package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue(){
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

1. 在consumer中编写消费逻辑，监听simple.queue。
  在consumer服务中编写application.yml，添加mq连接信息

spring:
  rabbitmq:
    host: 192.168.127.131 #rabbitMQ的IP地址
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: /

在consumer服务中新建一个类，编写消费逻辑

package cn.itcast.mq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SpringRabbitListener {


    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(String msg){
        System.out.println("消费者接收到simple.queue的消息是：【" + msg + "】");
    }
}

3.Work Queue工作队列

基本思路：

1.在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue
2.在consumer服务中定义两个消息监听者，每秒都监听simple.queue队列
3.消费者1每秒处理50条消息，消费者2处理10条消息

步骤：

// publisher服务 代码
package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello, message__";
        for (int i = 1; i <= 50; i++) {
            rabbitTemplate.convertAndSend(queueName, message + i);
            Thread.sleep(20);
        }
    }
}

// consumer服务代码
package cn.itcast.mq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到消息是：【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2接收到消息是：【" + msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }
}

上面代码无法实现预取效果，消息会被两个消费者平均分配
消息预取限制 修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限。

spring:
  rabbitmq:
    host: 192.168.127.131 #rabbitMQ的IP地址
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: /
    listener:
      simple:
        prefetch: 1 # 每次只获取1条消息

五 Work Queue工作队列

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。
常见exchange类型包括：

Fanout：广播
Direct：路由
Topic：话题

发布订阅模型

注意：exchange负责消息路由，而不是存储，路由失败则消息丢失

2.发布订阅 FanoutExchange

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

实现思路：

1.在consumer服务中，利用代码声明队列、交换机、并将两者绑定
2.在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2
3.在publisher服务中编写测试方法，向itcast.fanout（交换机）发送消息

步骤：

1.在consumer服务服务声明Exchange、Queue、Binding

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

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FaboutConfig {
    // itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    // fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

//    绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }

    // fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }
}

2.在consumer服务声明两个消费者

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

package cn.itcast.mq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "fanout.queue")
    public void listenFanoutQueue1(String msg){
        System.out.println("消费者接收到fanout.queue1的消息是：【" + msg + "】");
    }

    @RabbitListener(queues = "fanout.queue")
    public void listenFanoutQueue2(String msg){
        System.out.println("消费者接收到fanout.queue2的消息是：【" + msg + "】");
    }
}

1. 在publisher服务发送消息到FanoutExchange

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

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendFanoutExchange() {
        // 交换机名称
        String exchangeName = "itcast.fanout";
        // 消息
        String message = "hello, every one!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "", message);
    }

}

交换机的作用是什么？

接收publisher发送的消息
将消息按照规则路由到与之绑定的队列
不能缓存消息，路由失败，消息丢失
FanoutExchange的会将消息路由到每个绑定的队列

2.发布订阅 DirectExchange

Direct Exchange会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式(routes)。

每一个Queue都与Exchange设置一个BindingKey
发布者发送消息时，指定消息的RoutingKey
Exchange将消息路由到BindingKey与消息RoutingKry一致的队列

实现思路：

1.利用@RabbitListener声明Exchange、Queue、RoutingKey
2.在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
3.在publisher服务中编写测试方法，向itcast.fanout（交换机）发送消息

步骤：

1.在consumer服务声明Exchange、Queue

(1)在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
(2)并利用@RabbitListener声明Exchange、Queue、RoutingKey

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;

@Component
public class SpringRabbitListener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息是【" + msg + "】");
    }

}

2.在publisher服务发送消息到DirectExchange

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

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Test
    public void testSendDirectExchange() {
        // 交换机名称
        String exchangeName = "itcast.direct";
        // 消息
        String message = "hello, every blue!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "blue", message);
    }
}

3.发布订阅 TopicExchange

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

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

#：指代0个或多个单词

*：指代一个单词

TopicExchange

实现思路：

1.利用@RabbitListener声明Exchange、Queue、RoutingKey
2.在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
3.在publisher服务中编写测试方法，向itcast.topic（交换机）发送消息

步骤：

1.在consumer服务声明Exchange、Queue

(1)在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
(2)并利用@RabbitListener声明Exchange、Queue、RoutingKey

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;

@Component
public class SpringRabbitListener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("topic.queue1"),
            exchange = @Exchange(value = "itcast.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenerTopicQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息是【" + msg + "】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("topic.queue2"),
            exchange = @Exchange(value = "itcast.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenerTopicQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息是【" + msg + "】");
    }
}

2.在publisher服务发送消息到TopicExchange

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

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Test
    public void testSendTopicExchange() {
        // 交换机名称
        String exchangeName = "itcast.topic";
        // 消息
        String message = "中国新闻！";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
    }

}

六 SpringAMQP消息转换器

案例测试发送Object类型消息

说明：在SpringAMQP的发送方法中。接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列号为字节后发送。

我们在consumer中利用@Bean声明一个队列

package cn.itcast.mq.config;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FaboutConfig {
    @Bean
    public Queue ObjectQueue(){
        return new Queue("object.queue");
    }
}

在publisher中发送消息以测试：

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendObjectQueue() {
        Map<String, Object> msg = new HashMap<>();
        msg.put("name", "柳岩");
        msg.put("age", 21);
        rabbitTemplate.convertAndSend("object.queue", msg);
    }
}

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。
而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。

如果要修改只需要定义一个MessageConverter类型的Bean即可。推荐用JSON方式序列化，步骤如下:

在publisher服务中引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

在publisher服务声明MessageConvert：

// 在配置类中
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

接收消息

1.我们在consumer服务中引入Jackson依赖:

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

2.在consumer服务声明MessageConvert：

// 在配置类中
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

3.监听消息

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {\

    @RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String, Object> msg) {
        System.out.println("接收到object.queue的消息【" + msg + "】");
    }
}

一 同步通信

1.同步通信的问题

2.同步调用的优点

二 异步调用

1.异步通信的优点

2.异步通信的缺点

三 RabbitMQ 快速入门

1.什么是MQ？

2.RabbitMQ 概述

3.单机部署RabbitMQ

4.RabbitMQ概述

5.RabbitMQ中的几个概念

6.常见消息模型（参考链接)

(1) HelloWorld案例

四 SpringAMQP

1.什么是SpringAMQP？

2.Basic Queue工作队列

3.Work Queue工作队列

五 Work Queue工作队列

1.发布(Publish)、订阅(Subscribe)

2.发布订阅 FanoutExchange

2.发布订阅 DirectExchange

3.发布订阅 TopicExchange

六 SpringAMQP消息转换器

一同步通信

二异步调用