消息中间件-RabbitMQ介绍

一、基础知识

 

1. 什么是RabbitMQ

RabbitMQ是2007年发布，是一个在AMQP(高级消息队列协议)基础上完成的，简称MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法，由Erlang（专门针对于大数据高并发的语言）语言开发，可复用的企业消息系统，是当前最主流的消息中间件之一，具有可靠性、灵活的路由、消息集群简单、队列高可用、多种协议的支持、管理界面、跟踪机制以及插件机制。

 

 

2.什么是消息和队列

1.消息 就是数据，增删改查的数据。例如在员工管理系统中增删改查的数据

 

2.队列 指的是一端进数据一端出数据，例如C#中(Queue数据结构）

 

 

3.什么是消息队列

1.消息队列指：一端进消息，一端出消息

 

2.RabbitMQ就是实现了消息队列概念的一个组件，以面向对象的思想去理解，消息队列就是类，而RabbitMQ就是实例，当然不仅仅只有RabbitMQ,例如ActiveMQ，RocketMQ，Kafka，包括Redis也可以实现消息队列。

 

 

4.什么地方使用RabbitMQ

1.在常见的单体架构中，主要流程是用户UI操作发起Http请求>服务器处理>然后由服务器直接和数据库交互，最后同步反馈用户结果

 

 

 

2.在微服务架构中，例如下图中的员工管理系统，UI与微服务通信，主要是通过Http或者gRPC同步通信

 

 

 

 

问题分析

 

在上述2种情况下，我们发现在UI请求时都是同步操作 ，第2种架构虽然将整体服务按业务拆分成不同的微服务并且对应各自的数据库，但是在用户与微服务通信时，存在的问题依然没有解决，例如数据库的承载能力只能处理10w个请求，如果遇到高并发情况下，UI发起50w请求，那数据库是远远承载不了的,从而导致如下问题。

 

1.高并发请求导致系统性能下降响应慢，同时数据库承载风险加大

 

2.扩展性不强UI操作的交互对业务的依赖较大，导致用户体验下降

 

3.瞬时流量涌入巨大的话，服务器可能直接挂了

 

 

解决方案

 

 

 

为了解决性能瓶颈问题。我们需要将同步通信换成异步通信方式。因此就使用消息队列，用户在UI中操作直接写入RabbitMQ然后直接返回，剩下的业务操作由消息队列和各自的微服务来完成

RabbitMQ的优势

 

异步处理，响应快，增加了数据库（服务器的承载能力）

 

削峰，可以把流量的高峰分解到不同的时间段来处理

 

解耦（扩展性就更强），让UI和业务独立演化

 

高可用，处理器如果发生故障了，对其他的处理器没有影响

 

RabbitMQ的不足

 

增加了系统复杂性，不方便调试和开发，在使用RabbitMQ以前前端直接和服务交互，现在加了一层

 

即时性降低了,在某一程度上提升了用户操作体验，也降低了用户体验，但是避免不了，取长补短

 

更加依赖消息队列了

 

 

5.RabbitMQ组成概念

1.ConnectionFactory 为Connection的制造工厂。

 

2.Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。

 

3.Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

 

4.Exchange（交换机） 我们通常认为生产者将消息投递到Queue中，实际上实际的情况是，生产者将消息发送到Exchange，由Exchange将消息路由到一个或多个Queue中（或者丢弃），而在RabbitMQ中的Exchange一共有4种策略，分别为：fanout（扇形）、direct（直连）、topic（主题）、headers（头部）

 

二、如何落地RabbitMQ

1.RabbitMQ环境安装

1.下载RabbitMQ

 

2.运行环境erlang

 

3.安装完成之后，加载RabbitMQ管理插件

 

rabbitmq-plugins enable rabbitmq_management

4.安装成功访问RabbitMQ管理后台http://localhost:15672

 

2.创建系统业务

1.分别创建考勤服务，请假服务，计算薪酬服务，邮件服务，短信服务消费者角色

 

2.创建员工管理网站用于模拟前端调用，主要充当生产者角色

 

3.在员工管理网站和每一个模拟微服务中通过nuget引入RabbitMQ.Client

 

4.在员工管理网站中创建模拟添加考勤的控制器并加入生产者代码

 

 //创建连接

 using (var connection = factory.CreateConnection())

 {

     //创建通道

     var channel = connection.CreateModel();

     //定义队列

     channel.QueueDeclare("CreateAttendance", false, false, false, null);

 

     string json = JsonConvert.SerializeObject(attendanceDto);

 

     //创建内容对象

     var properties = channel.CreateBasicProperties();

     //发送消息

     channel.BasicPublish(exchange: "",routingKey: "CreateAttendance",basicProperties: properties,body: Encoding.UTF8.GetBytes(json));

 }

 

5.在考勤微服务中创建接口，并在接口中加入消费者代码

 

var connection = factory.CreateConnection();

var channel = connection.CreateModel();   

//创建消费者事件

var consumer = new EventingBasicConsumer(channel);

consumer.Received += (model, ea) =>

{

    var body = ea.Body;

    // 1、逻辑代码，添加到数据库

    var message = Encoding.UTF8.GetString(body.ToArray());

    object json = JsonConvert.DeserializeObject(message);

    Console.WriteLine(" [x] 创建考勤信息 {0}", message);

};

//设置消费者属性

//p1.监听队列p2.消息确认ACK p3.消费者实例赋值

channel.BasicConsume(queue: "CreateAttendance",autoAck: false,consumer:consumer);