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

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

1、消息中间件

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

整体来讲，消息中间件有如下做用:下降耦合、流量消峰（防浪涌）、可靠性传输、事件驱动

1.下降耦合:经过发布订阅的方式松耦合

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

邮件业务和短信业务的代码是写在用户注册的流程里，不管是经过接口的方式来实现，仍是远程调用的方式来实现，耦合度都很高，如今，新增一个需求，用户注册完成之后不发送邮件了，而是给用户“增长积分”，咱们来分析这几种状况：

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

第2、经过远程调用的方式，代码相似以下

当咱们要新增业务处理时，以下

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

若是咱们引入消息中间件，以下：

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

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

3.流量消峰、防浪涌:消息中间件能够在峰值时堆积消息，而在峰值过去后下游系统慢慢消费消息解决流量洪峰的问题

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

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

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

5.事件驱动：系统与系统之间能够经过消息传递的形式驱动业务

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

6.分布式事务支持

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

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

2、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的架构图以下(来自网络)：

3、集群部署和测试

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伪集群搭建完毕。

未完待续。。。