同步和异步通讯

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

同步通讯

同步调用的优点：

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

同步调用的问题：

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

异步通讯

通过Broker实现,Broker 是一个像数据总线一样的东西，所有的服务要接收数据和发送数据都发到这个总线上，这个总线就像协议一样，让服务间的通讯变得标准和可控。

好处：

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

缺点：

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

MQ

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

比较常见的MQ实现：

ActiveMQ
RabbitMQ
RocketMQ
Kafka
几种常见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

Rabbit

centos7 docker安装

拉取镜像
docker pull rabbitmq:management
创建容器并启动
docker run -id –name=rabbit –hostname=rabbit -p 15672:15672 -p 5672:5672 rabbitmq:management

RabbitMQ消息模型(队列消息发布/订阅消息)

RabbitMQ中的几个概念：

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

队列消息

特点：

不会丢失数据且有序（先进先出）
仅有一个消费者能拿到
- 模型：
1. SimpleQueue（简单）
  1vs1 队列中待处理的消息少
2. WorkQueue （工作）
  1vsN 队列中待处理的消息多,多个消费会平均分配,导致性能浪费,设置平均分配为能者多劳模式

订阅消息

特点：

消息会丢失
强调的是多个消费者必须同时在线才能同时收到消息,不然消息会丢失
必须要有交换
- 模型：
1. fanout:
  1. 通过队列绑定交换机
  2. 数据通过交换机发给绑定的队列,所有消费者都能收到数据
2. direct:在fanout的基础上可以通过routingkey(路由键)进行分类,指定了路由键的消费者才能收到数据
3. topic:在direct的基础上可以对routingkey(路由键)进行通配简化
  1. *:可以匹配一个词 #:可以匹配多个词
  2. 指定了通配路由键的消费者才能收到数据

java复制代码public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        //1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        //1.1设置连接参数,分别是:主机名,端口号、vhost、用户名、密码
        factory.setHost("192.168.190.131");
        factory.setPort(5672);
        factory.setVirtualHost("hawk-house");
        factory.setUsername("hawk");
        factory.setPassword("123456");
        //1.2建立连接
        Connection connection = factory.newConnection();
        //2创建通道Channel
        Channel channel = connection.createChannel();
        //3创建队列queue
        String queueName = "simple.queue";
        channel.queueDeclare(queueName,false,false,false,null);
        //4发送消息
        String message = "hello,dover";
        channel.basicPublish("",queueName,null,message.getBytes());
        System.out.println("发送消息成功：【" + message + "】");
        //5关闭通道和连接
        channel.close();
        connection.close();
    }
}

java复制代码public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        connectionFactory.setHost("192.168.190.131");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("hawk-house");
        connectionFactory.setUsername("hawk");
        connectionFactory.setPassword("123456");
        // 1.2.建立连接
        Connection connection = connectionFactory.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("等待接收消息。。。。");
    }
}

SpringAMQP

什么是AMQP？
应用间消息通信的一种协议，与语言和平台无关。

SimpleQueue

消息发送

application.yml中添加配置

yml复制代码spring:
  rabbitmq:
    host: 192.168.190.131
    port: 5672
    virtual-host: hawk-house
    username: hawk
    password: 123456

利用RabbitTemplate实现消息发送：

java复制代码@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitTemplateTest {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    public void SingleAMOP(){
        String queueName="simple.queue";
        String message="hello,hawk";
        rabbitTemplate.convertAndSend(queueName,message);
    }
}

消息接收

application.yml中添加配置:同发送
自定义一个消息监听器

此类交给容器管理
编写方法：监听消息队列（队列中一旦有最新的消息，自动执行方法）
- 方法没有返回值
- 方法有参数（参数=发送的消息内容）
- 参数类型和发送的消息类型一致
- 在方法上通过@RabbitListener绑定队列

java复制代码@Component
public class MessageListener {

    @RabbitListener(queues = "simple.queue")
    public void SingleQueueGet(String message) throws InterruptedException {
        
        System.out.println(message);
    }
}

WorkQueue

Work queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度,此时就可以使用work模型

消息发送

java复制代码@Test
public void sendWorkQueue(){
    String queueName="simple.queue";
    for (int i = 0; i < 50; i++) {
        String message="hello,hawk"+"-"+i;
        rabbitTemplate.convertAndSend(queueName,message);
    }

}

消息接收

默认多个消费者轮询接收信息

java复制代码@Component
public class MessageListener {

    @RabbitListener(queues = "simple.queue")
    public void SingleQueueGet(String message) throws InterruptedException {

        System.out.println("消费者1:"+message);
    }
}

java复制代码@Component
public class MessageListener {

    @RabbitListener(queues = "simple.queue")
    public void SingleQueueGet(String message) throws InterruptedException {
        Thread.sleep(1000);
        System.out.println("消费者2:"+message);
    }
}

能者多劳

application.yml文件，添加配置,注意所有消费者都要添加配置,只添加一个消费者的话该消费者处理一条信息后就会停止获取信息

yml复制代码spring:
  rabbitmq:
    listener:  #将workqueue的默认平均分配处理改成能者多捞处理
      simple:
        prefetch: 1

发布/订阅

在订阅模型中，多了一个exchange角色，过程略有变化：

生产者：也就是要发送消息的程序，但是不再发送到队列中，而是发给交换机
Exchange：交换机.一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列
- Direct：定向，把消息交给符合指定routing key 的队列
- Topic：通配符，把消息交给符合routing pattern（路由模式）的队列
消费者：与以前一样，订阅队列，没有变化
Queue：消息队列也与以前一样，接收消息、缓存消息。

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

Fanout

消息发送流程:

1）可以有多个队列
2）每个队列都要绑定到Exchange（交换机）
3）生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
4）交换机把消息发送给绑定过的所有队列
5）订阅队列的消费者都能拿到消息

消息发送

java复制代码@Test
public void Fanout(){
    //交换机名称
    String exchangeName="hawk-ex";
    String message="hi,dover";
    /*数据通过交换机给队列
             交换机名称
             路由键（路由模型下用）
             数据*/
    rabbitTemplate.convertAndSend(exchangeName,"",message);
}

消息接收

java复制代码@RabbitListener(
        //绑定队列和交换机
        bindings = @QueueBinding(
                //指定队列
                value = @Queue(
                        //队列名称
                        value ="fanout.queue1"
                ),
                //指定交换机
                exchange = @Exchange(
                        //交换机名称
                        value = "hawk-ex",
                        //交换机类型
                        type = ExchangeTypes.FANOUT
                )
        ))
public void Fanout(String message) {
    System.out.println("消费者1:"+message);
}

Direct

不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下：

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

消息发送

java复制代码@Test
public void Direct(){
    String exchangeName="hawk-dir";
    String routingKey="9812";
    String message="hawk,hawk,hawk";
    rabbitTemplate.convertAndSend(exchangeName,routingKey,message);
}

消息接受

java复制代码@RabbitListener(
        //绑定队列和交换机
        bindings = @QueueBinding(
                //指定队列
                value = @Queue(
                        //队列名称
                        value ="direct.queue1"
                ),
                //指定交换机
                exchange = @Exchange(
                        //交换机名称
                        value = "hawk-dir",
                        //交换机类型
                        type = ExchangeTypes.DIRECT
                ),
                key = {"0630","9812","0304"}
        ))
public void Direct(String message) {
    System.out.println("消费者1:"+message);
}

Topic

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

消息发送

java复制代码@Test
public void topic(){
    String exchangeName="hawk-dir";
    String routingKey="hawk.dover.9999";
    String message="hawk,hawk,hawk";
    rabbitTemplate.convertAndSend(exchangeName,routingKey,message);
}

消息接收

java复制代码@RabbitListener(
        //绑定队列和交换机
        bindings = @QueueBinding(
                //指定队列
                value = @Queue(
                        //队列名称
                        value ="topic.queue1"
                ),
                //指定交换机
                exchange = @Exchange(
                        //交换机名称
                        value = "hawk-topic",
                        //交换机类型
                        type = ExchangeTypes.TOPIC
                ),
                key = "hawk.#"
        ))
public void topic(String message) {
    System.out.println("消费者1:"+message);
}

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

同步和异步通讯

同步通讯

异步通讯

MQ

Rabbit

centos7 docker安装

RabbitMQ消息模型(队列消息 发布/订阅消息)

队列消息

订阅消息

SpringAMQP

SimpleQueue

消息发送

消息接收

WorkQueue

消息发送

消息接收

能者多劳

发布/订阅

Fanout

消息发送

消息接收

Direct

消息发送

消息接受

Topic

消息发送

消息接收

RabbitMQ消息模型(队列消息发布/订阅消息)