微服务与消息驱动(RabbitMQ,Kafka)之入门篇

清疚 2022-03-08 06:36 219阅读 0赞

本篇来自于个人阅读书籍后的总结--有误之处还请大家斧正,谢谢!

## **Spring Could Stream 介绍** ##

**1.关于****Stream框架**

首先我们为什么要先了解Spring could Stream,它是一个用户构建消息驱动微服务的框架,框架在Spring boot 基础上整合了spring Integration 来连接消息代理中间件.支持多个消息中间件的自定义配置,同时吸收了这些消息中间件的部分概念,如 持久化订阅,消费者分组和分区等等,

**目前 Spring could Stream 的绑定器提供 RabbitMQ与Kafka消息中间件的实现**

使用**Stream 框架**我们不必关系如何连接各个消息代理中间件,也不必关心如何进行消息的发送与接收,只需简单配置就可以实现这些功能使我们可以更加专注于业务逻辑的开发.

**2.Stream 框架的组成部分**

*  Stream 框架应用模型
 *  绑定器抽象层,可以与消息代理中间件进行绑定,通过绑定器的API可实现插件式的绑定器
 *  持久化订阅支持
 *  消费者组的支持
 *  Topic分区支持

**3.消息代理中间件(ActiveMQ,RabbitMQ,Kafka)**

调用他们的API用来发送,接收消息, 大概如下图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70][]

Spring could Stream 在生产与消费者之间加入了一个类似代理的角色直接与消息代理进行交互,使消息生产者与消费者不在需要直接调用各种消息代理框架的API 甚至感受不到消息代理的存在

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 1][]

## RabbitMQ 框架 ##

**1. Rabbit MQ 和 AMQP**

Rabbit MQ 是一个轻量级的消息代理中间件,支持多种消息通信协议,分布式部署,同时支持运行于多个操作系统,具有灵活 高可用等特性,因此 Rabbit MQ广受欢迎

Rabbit MQ支持多种协议最为重要的是高级消息列队协议(**AMQP**),它定义了"消息客户端"与"消息中间件"之间的通信协议,基于该协议 "消息客户端"与"消息代理中间件" 可以不受开发语言,具体产品的约束

** AMQP大致模型:**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 2][]

**如图,生产者会将消息投递给 消息代理(**Rabbit MQ服务器**),他们之间会建立消息通道(Channel),消息由交换器(Exchange)先进性处理,交换器会选择把消息交给哪一个列队(Message Queue),最后消息队列会将消息发送消费者**

**1. Rabbit MQ 下载与安装**

**Rabbit MQ 下载: **[http://www.rabbitmq.com/download.html][http_www.rabbitmq.com_download.html]

**Rabbit MQ服务器使用的是Erlang语言因此还要下载Erlang :**[http://erlang.org/download/otp\_win64\_21.3.exe][http_erlang.org_download_otp_win64_21.3.exe]

**  注意:下载Rabbit MQ和Erlang安装请注意,若操作系统是64位请统一下载64位版本安装否则使用时会有异常**

*     依次安装Erlang和RabbitMQ 命令行进入RabbitMQ的sbin目录
 *     输入: rabbitmq-plugins list  查看当前插件状态
 *     使用控制台需开启插件管理: rabbitmq-plugins enable rabbitmq\_management
 *     浏览器打开 http://localhost:15672 可以看到RabbitMQ登录界面
 *     默认的用户名和密码均为 guest

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 3][]

**2. 编写生产者与消费者**

**   对于服务器来说 消息生产者与消费者都属于客户端,他们与通过服务器之间通过AMQP协议进行通信,不受语言限制**

**   引入jar**

<dependency>
    			<groupId>com.rabbitmq</groupId>
    			<artifactId>amqp-client</artifactId>
    			<version>4.2.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.slf4j</groupId>
    			<artifactId>slf4j-log4j12</artifactId>
    			<version>1.7.9</version>
    		</dependency>

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 4][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 5][]

主要过程可以在**Rabbit MQ **的主界面查看到(执行一步查阅一下)

**3. 交换器 ,绑定 与 列队**

**   RabbitMQ在设计上,生产者不会直接将消息发给队列而是发给交换器,交换器一边接收生产者消息一边将消息发送给各个队列,**

**交换器与队列之间的连接叫做 绑定(binding). 每个绑定都有一个名字叫做 binding key,消息发过来时会带有一个路由的标识(routing key). 交换器会根据这两个值来决定将消息发送给哪个队列.交换器实际上也充当着路由的作用**

**  RabbitMQ 提供4中类型的交换器**

*  **direct :   根据生产者传过来的 routing key是否等于 binding key来决定将消息发送给哪个队列**
 *  **topic:     根据生产者传过来的 routing key是否匹配一定的表达式来决定将消息发送给哪个队列**
 *  **fanout:   将消息发送给交换器知道的全部队列,忽略设置的 routing key**
 *  **headers:根据消息头来决定将消息发送给哪个队列**

## Apache Kafka 框架 ##

**1.关于  Kafka**

**Kafka 是**Apache 下的一个用于处理数据流的分布式消息框架,拥有水平扩展,容错,高效等特性,可以实现以下功能:

*  构建在系统间进行实时数据传输的通道
 *  构建对数据流进行转换或者响应的实时应用

kafka的整体结构与**RabbitMQ** 类似,生产者的消息会被发送到Topic中,topic保存着各类数据,每一条数据都使用键值进行保存.每一个topic中都包含一个或多个物理分区(partition),这些分区维护着消息的内容和索引,他们有可能被保存在不同服务器中,对于客户端来说无须关心数据如何被保存,只需关系将消息发往那个topic.

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 6][]

**2.运行  Kafka服务器**

*  kafka依赖了 Zookeeper, 启动 kafka服务器前需要先启动Zookeeper
 *  下载kafka与Zookeeper,下载后解压
 *  进入zookeeper-3.4.13/conf目录,将zoo\_sample.cfg文件复制一份,并重命名zoo.cfg
 *  命令行进入zookeeper-3.4.13/bin目录,运行zkServer命令,默认端口:2181,窗口不必关接下来启动kafka
 *  命令行进入kafka\_2.11-2.1.1/bin/windows目录,运行kafka-server-start ../../config/server.properties命令启动kafka服务,默认端口:9092

**2.编写生产者和消费者与消费者组**

**引入jar包**

<dependency>
    			<groupId>org.apache.kafka</groupId>
    			<artifactId>kafka-clients</artifactId>
    			<version>0.11.0.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.slf4j</groupId>
    			<artifactId>slf4j-log4j12</artifactId>
    			<version>1.7.9</version>
    		</dependency>

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 7][]

命令行进入kafka\_2.11-2.1.1/bin/windows目录,运行kafka-topics --list --zookeeper localhost:2181 查看kafka服务器的Topic

删除服务器Topic: kafka-topics --delete --zookeeper localhost:2181 --topic my-topic  ,默认情况下该命令只删除topic标记,如果想真实删除需修改 config/server.properties文件,加入delete.topic.enable=true 配置.

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 8][]

**   消费者组:在编写消费者时需要指定消费者组的id,消费者会为自己添加一个消费者组的标识,每一条发布到topic的记录都会被交付给消费者组的一个消费者实例.如果多个消费者实例拥有相同的消费者组,那么这些记录将会均衡分配到各个消费者实例已达到负载均衡的目的(如果A和B为同一个消费者组,生产者发送一条消息仅会交给其中一个消费者).  如果所有的消费者都不同的消费者组,那么每一条记录都会广播到全部的消费者进行处理(A和B为不同消费者组,生产者发送一条消息A,B都会收到进行处理)**

![20190315104257883.png][]

以上是概述RabbitMQ和Kafka的入门, 下篇将更新**Spring could Stream集成 **RabbitMQ和Kafka 开发消息微服务!

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70]: /images/20220308/85219f0b0f2546ab8bfde0612fd6ee14.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 1]: /images/20220308/2d2af10253be4a0895742550591bb772.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 2]: /images/20220308/0748ce0c69774a22939a49173dcac517.png
[http_www.rabbitmq.com_download.html]: http://www.rabbitmq.com/download.html
[http_erlang.org_download_otp_win64_21.3.exe]: http://erlang.org/download/otp_win64_21.3.exe
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 3]: /images/20220308/212b0db8d5c84a0b86e2d11b0bb4407d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 4]: /images/20220308/b8c975c90d3a47efaf02531d788d3311.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 5]: /images/20220308/883abb1c561d427e8e2281ecbbe2691a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 6]: /images/20220308/eb138ec379104585bd8de47846af840c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 7]: /images/20220308/5ad8b881dbc44c5a87495e476dc2e669.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xjaHExOTk1_size_16_color_FFFFFF_t_70 8]: /images/20220308/87ba2ffa136f446d9eb131e771ae4dc6.png
[20190315104257883.png]: /images/20220308/efea8d3d1c864d4d9927ca6078edfc3c.png