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

本文涉及的产品
Serverless 应用引擎 SAE,800核*时 1600GiB*时
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
云原生网关 MSE Higress,422元/月
简介: 业务系统中,通常会遇到这些场景:A系统向B系统主动推送一个处理请求;A系统向B系统发送一个业务处理请求,因为某些原因(断电、宕机。。),B业务系统挂机了,A系统发起的请求处理失败;前端应用并发量过大,部分请求丢失或后端业务系统卡死。

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

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

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

1.降低耦合:通过发布订阅的方式松耦合
我们以注册业务为例,注册成功会发送短信、邮件给用户来确认,传统架构模型是这样:

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

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

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

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

还是要改动主流程代码,避免不了启停应用。
如果我们引入消息中间件,如下:

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

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

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,详细配置如下:

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=/usr/zookeeper/zookeeper0/zkdata
dataLogDir=/usr/zookeeper/zookeeper0/zkdatalog
# the port at which the clients will connect
clientPort=4180
server.0=192.168.217.129:8880:7770    
server.1=192.168.217.129:8881:7771    
server.2=192.168.217.129:8882:7772 
#server.0 这个0是服务器的标识也可以是其他的数字, 表示这个是第几号服务器,用来标识服务器,这个标识要写到快照目录下面myid文件里
#192.168.217.129为集群里的IP地址,第一个端口是master和slave之间的通信端口

以下是配置文件的解释:

#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

#server0
echo "0" > /usr/zookeeper/zookeeper0/zkdata
#server2
echo "1" > /usr/zookeeper/zookeeper1/zkdata
#server3
echo "2" > /usr/zookeeper/zookeeper2/zkdata

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

启动zookeeper
进入bin目录;执行 ./zkServer.sh ../conf/zoo.cfg命令

检查启动状态:
./zkServer.sh status

mode:leader 表示这是个leader节点
其他两个实例同样操作。至此zookeeper伪集群搭建完毕。
本文地址https://my.oschina.net/u/3112259/blog/1556837

相关文章
|
18天前
|
消息中间件 存储 中间件
【消息中间件】详解三大MQ:RabbitMQ、RocketMQ、Kafka
【消息中间件】详解三大MQ:RabbitMQ、RocketMQ、Kafka
45 0
|
2月前
|
消息中间件 分布式计算 监控
Python面试:消息队列(RabbitMQ、Kafka)基础知识与应用
【4月更文挑战第18天】本文探讨了Python面试中RabbitMQ与Kafka的常见问题和易错点,包括两者的基础概念、特性对比、Python客户端使用、消息队列应用场景及消息可靠性保证。重点讲解了消息丢失与重复的避免策略,并提供了实战代码示例,帮助读者提升在分布式系统中使用消息队列的能力。
75 2
|
2月前
|
消息中间件 监控 固态存储
性能工具之 Kafka 快速 BenchMark 测试示例
【5月更文挑战第24天】性能工具之 Kafka 快速 BenchMark 测试示例
54 1
性能工具之 Kafka 快速 BenchMark 测试示例
|
3天前
|
消息中间件 Java Kafka
Spring Boot与Kafka的集成应用
Spring Boot与Kafka的集成应用
|
5天前
|
消息中间件 存储 大数据
深度分析:Apache Kafka及其在大数据处理中的应用
Apache Kafka是高吞吐、低延迟的分布式流处理平台,常用于实时数据流、日志收集和事件驱动架构。与RabbitMQ(吞吐量有限)、Pulsar(多租户支持但生态系统小)和Amazon Kinesis(托管服务,成本高)对比,Kafka在高吞吐和持久化上有优势。适用场景包括实时处理、数据集成、日志收集和消息传递。选型需考虑吞吐延迟、持久化、协议支持等因素,使用时注意资源配置、数据管理、监控及安全性。
|
2月前
|
消息中间件 存储 传感器
Kafka消息队列原理及应用详解
【5月更文挑战第6天】Apache Kafka是高性能的分布式消息队列,常用于实时数据管道和流应用。它提供高性能、持久化、分布式和可伸缩的消息处理,支持解耦、异步通信和流量控制。Kafka的核心概念包括Broker、Topic、Partition、Producer、Consumer和Consumer Group。其特点是高吞吐、低延迟、数据持久化、分布式架构和容错性。常见应用包括实时数据流处理、日志收集、消息传递和系统间数据交换。
|
2月前
|
消息中间件 Oracle 关系型数据库
实时计算 Flink版操作报错之连接外部kafka本地执行测试代码报错如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
2月前
|
消息中间件 Kafka
Kafka【部署 03】Zookeeper与Kafka自动部署脚本
【4月更文挑战第11天】Kafka【部署 03】Zookeeper与Kafka自动部署脚本
38 8
|
2月前
|
消息中间件 存储 数据管理
【Kafka】zookeeper对于kafka的作用是什么?
【4月更文挑战第6天】【Kafka】zookeeper对于kafka的作用是什么?
|
2月前
|
消息中间件 Kafka Linux
Kafka【付诸实践 03】Offset Explorer Kafka 的终极 UI 工具安装+简单上手+关键特性测试(一篇学会使用 Offset Explorer)
【2月更文挑战第21天】Kafka【付诸实践 03】Offset Explorer Kafka 的终极 UI 工具安装+简单上手+关键特性测试(一篇学会使用 Offset Explorer)
445 2

相关产品

  • 云消息队列 Kafka 版
  • 微服务引擎