消息中间件 kafka+zookeeper 集群部署测试与应用

Posted 2021-04-13 OSC开源社区

扫描二维码或点击图片进入西安源创会报名

业务系统中，通常会遇到这些场景：A系统向B系统主动推送一个处理请求；A系统向B系统发送一个业务处理请求，因为某些原因（断电、宕机。。），B业务系统挂机了，A系统发起的请求处理失败；前端应用并发量过大，部分请求丢失或后端业务系统卡死。。。。这个时候，消息中间件就派上用场了--提升系统稳定性、可用性、可扩展性。

一、消息中间件

消息队列技术是分布式应用间交换信息的一种技术。消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读走。通过消息队列，应用程序可独立地执行--它们不需要知道彼此的位置、或在继续执行前不需要等待接收程序接收此消息。

总体来说，消息中间件有以下作用:降低耦合、流量消峰（防浪涌）、可靠性传输、事件驱动

1.降低耦合:通过发布订阅的方式松耦合

我们以注册业务为例，注册成功会发送短信、邮件给用户来确认，传统架构模型是这样：

邮件业务和短信业务的代码是写在用户注册的流程里，无论是通过接口的方式来实现，还是远程调用的方式来实现，耦合度都很高，现在，新增一个需求，用户注册完成以后不发送邮件了，而是给用户“增加积分”，我们来分析这几种情况：

第一、都在一个业务系统内通过代码堆积、接口调用的方式来实现注册成功后的业务处理，我们需要改动注册代码，上线时需要启停应用，这种方式耦合度最高。

第二、通过远程调用的方式，代码类似如下

当我们要新增业务处理时，如下

还是要改动主流程代码，避免不了启停应用。

如果我们引入消息中间件，如下：

通过发布订阅的方式，用户注册成功后，给消息中间件发送一个消息，各个业务端订阅同一个频道的消息，接收到注册成功的消息后就执行相应业务。新增“增加积分”，就让积分系统相关处理订阅这个消息就够了，不用改动用户业务代码，不用启停应用，就可以实现业务的平滑扩展。

CRM中，这种应用场景有很多，再比如，用户回访业务，回访成功后发短信追评、记录历史工单等等，都可以使用消息中间件来送耦合，实现业务的平滑扩展。

2 和 3.流量消峰、防浪涌

消息中间件可以在峰值时堆积消息，而在峰值过去后下游系统慢慢消费消息解决流量洪峰的问题

并发量达到高峰时，后端系统压力通常会很大，无论是应用服务器还是数据库服务器，这个时候可以将类似请求放到消息中间件的消息队列中堆积起来慢慢处理，也可以设置消息队列的大小达到控制并发量的目的，商品秒杀的业务场景比较典型，在实际工作中遇到的业务场景也有很多，不一一列举。

4.可靠性传输

基于消息的通信是可靠的，消息不会丢失。大多数消息中间件都提供将消息持久化到磁盘的功能。开篇的第一个场景中，消息中间件可以将生产者生产的消息持久化到磁盘，后端应用宕机，但消息不会丢失，应用重启继续处理。

你可能会有这样的疑问，那消息中间件宕机了，接收不到消息怎么持久化？没关系，消息中间件提供了多种消息复制策略、持久化策略，集群部署策略等都可以保证消息的可靠性传输。那如果你是个爱钻牛角尖的人，你也许会问，集群中所有的节点都挂了怎么办。。。那如果世界末日了你还在care消息不消息的事情的话只能说明你足够敬业。

5.事件驱动

系统与系统之间可以通过消息传递的形式驱动业务

可以将复杂的应用系统重构成为事件驱动的系统。事件溯源（Event Sourcing），表示一个对象从创建到消亡，会经过的多种状态。如果把对象的状态变化都存储下来，不但可以根据状态变化记录获取对象的当前状态，也可以回溯对象的变化过程。消息中间件能很好地支持这样的系统设计方式，将触发对象状态变化的事件放入消息队列。

6.分布式事务支持

这个老生常谈的问题了，如果你不明白什么是分布式事务，请移步www.baidu.com。

kafka提供了对分布式事务的支持。欲知如何支持，且听下回分解。

二、kafka介绍

 

Kafka是由LinkedIn开发的一个分布式的消息系统，使用Scala编写，它以可水平扩展和高吞吐率而被广泛使用。目前越来越多的开源分布式处理系统如Cloudera、Apache Storm、Spark都支持与Kafka集成。淘宝的两款消息中间件，notify 和最近被Apache 加冕为顶级开源项目的RocketMQ都有kafka的影子。

总之，它很牛逼。

Kafka 专用术语：

• Broker：Kafka 集群包含一个或多个服务器，这种服务器被称为 broker。

• Topic：每条发布到 Kafka 集群的消息都有一个类别，这个类别被称为 Topic。（物理上不同 Topic 的消息分开存储，逻辑上一个 Topic 的消息虽然保存于一个或多个 broker 上，但用户只需指定消息的 Topic 即可生产或消费数据而不必关心数据存于何处）。

• Partition：Partition 是物理上的概念，每个 Topic 包含一个或多个 Partition。

• Producer：负责发布消息到 Kafka broker。

• Consumer：消息消费者，向 Kafka broker 读取消息的客户端。

• Consumer Group：每个 Consumer 属于一个特定的 Consumer Group（可为每个 Consumer 指定 group name，若不指定 group name 则属于默认的 group）

kafka的架构图如下(来自网络)：

三、集群部署和测试

kafka需要一个服务注册中心，你可以选择kafka自带的zookeeper，也可以单独部署。为了更好的管理，笔者单独部署zookeeper集群--其实是个伪集群，一台liux主机开放了三个端口即可。

zookeeper的运行需要jdk，所以先下载安装jdk，本文使用的jdk1.8

1.）下载安装jdk

jdk的安装步骤不做赘述， 笔者的版本是 jdk-8u144-linux-x64.tar.gz

配置环境变量（不会找度娘）

2.）下载zookeeper

先关闭防火墙，或者开放下文中需要用到的所有端口。

移步阿帕奇官网下载最新版zookeeper，

Zookeeper集群的工作是超过半数才能对外提供服务，3台中超过两台超过半数，允许1台挂掉。

创建目录 zookeeper ：

mkdir zookeeper

解压下载后的压缩文件 tar -zxvf zookeeper-3.4.6.tar.gz 

拷贝三份：

cp -r zookeeper-3.4.6 zookeeper0

cp -r zookeeper-3.4.6 zookeeper1

cp -r zookeeper-3.4.6 zookeeper2

进入zookeeper目录，其他的两个zookeeper实例同样操作:

创建两个目录zkdata、zkdatalog

进入conf目录，

拷贝zoo_sample.cfg 为zoo.cfg，详细配置如下：

以下是配置文件的解释：

#tickTime：

这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。

#initLimit：

这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。

当已经超过 5个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒

#syncLimit：

这个配置项标识 Leader 与Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是5*2000=10秒

#dataDir：

快照日志的存储路径

#dataLogDir：

事物日志的存储路径，如果不配置这个那么事物日志会默认存储到dataDir制定的目录，这样会严重影响zk的性能，当zk吞吐量较大的时候，产生的事物日志、快照日志太多

#clientPort：

这个端口就是客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求。

创建myid

其他两个实例修改端口，同样操作。

 

启动zookeeper

进入bin目录；执行 ./zkServer.sh ../conf/zoo.cfg命令

检查启动状态：

./zkServer.sh status

 

mode：leader 表示这是个leader节点

其他两个实例同样操作。至此zookeeper伪集群搭建完毕。

未完待续。。。

推荐阅读

点击“阅读原文”查看更多精彩内容