Kafka相关内容总结（Kafka集群搭建手记）

Kafka is a distributed,partitioned,replicated commit logservice。它提供了类似于JMS的特性，但是在设计实现上完全不同，此外它并不是JMS规范的实现。kafka对消息保存时根据Topic进行归类，发送消息者成为Producer,消息接受者成为Consumer,此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。无论是kafka集群，还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。

入门请参照： https://www.ibm.com/developerworks/cn/opensource/os-cn-kafka/index.html

在此不再赘述。

这部分不是本文的重点，但是kafka需要用到kafka集群，所以先搭建kafka集群。

从kafka官方文档看到，kafka似乎在未来的版本希望抛弃zookeep集群，自己维护集群的一致性，拭目以待吧。

我们搭建集群使用的是三台同机房的机器，因为zookeeper不怎么占资源也不怎么占空间（我们的业务目前比较简单），所以三台机器上都搭建了zookeeper集群。

搭建zookeeper集群没什么难度，参考文档： http://www.cnblogs.com/huangxincheng/p/5654170.html

下面列一下我的配置并解析：

一共用三台物理机器，搭建一个Kafka集群。

每台服务器的硬盘划分都是一样的，每个独立的物理磁盘挂在一个单独的分区里面，这样很方便用于Kafka多个partition的数据读写与冗余。

/data1比较小，为了不成为集群的瓶颈，所以/data1用于存放kafka以及Zookeeper

每台机器的磁盘分布如下：

下面是kafka的简单配置，三台服务器都一样，如有不一致的在下文有说明。

kafka安装在目录/usr/local/kafka/下，下面的说明以10.1.xxx.57为例。

最重要的配置文件server.properties，需要配置的信息如下：

从上面的配置看到，kafka集群不需要像hadoop集群那样，配置ssh通讯，而且一个kafka服务器（官方文档称之为broker，下面统一使用这个称呼）并不知道其他的kafka服务器的存在，因此你需要逐个broker去启动kafka。各个broker根据自己的配置，会自动去配置文件上的zk服务器报到，这就是一个有zk服务器粘合起来的kafka集群。

我写了一个启动脚本，放在 /usr/local/kafka/bin 下面。启动脚本每个broker都一样：

如同kafka集群里面每一个broker都需要单独启动一样，kafka集群里面每一个broker都需要单独关闭。

官方给出的关闭脚本是单独运行 bin/kafka-server-stop.sh

但是我运行的结果是无法关闭。打开脚本一看，才发现是最简单的办法，发一个TERM信号到kafka的java进程，官方脚本给出的grep有点问题。

发信号之后，一直tail着kafka日志，看到正常关闭。

指定zookeeper服务器，topic名称是LvsKafka（注意topic名称不能有英文句号(.)和下划线(_)，否则会通不过，理由是名称会冲突，下文对此略有解析）

replication-factor指出重复因子是2，也就是每条数据有两个拷贝，可靠性考虑。

partitions 指出需要多少个partition，数据量大的多一点，无论生产和消费，这是负载均衡和高并发的需要。

可以看到刚才新建的24个partition，比如partition 5， 他的leader是broker 59，也就是10.1.xxx.59这台机器。

建立topic时我们指出需要2个拷贝，从上面的输出的Replicas字段看到，这两个拷贝放在59,58两个机器，也就是10.1.xxx.59和10.1.xxx.58.

Isr表示当前partition的所有拷贝所在的机器中，哪些是还活着（可以提供服务）的。现在是59和58都还存活。

这个命令另外还会看到一些类似于下面的内容：

__consumer_offsets到底是什么呢？其实就是客户端的消费进度，客户端会定时上报到kafka集群，而kafka集群会把每个客户端的消费进度放入一个自己内部的topic中，这个topic就是__consumer_offsets。我查看过__consumer_offsets的内容，其实就是每个客户端的消费进度作为一条消息，放入__consumer_offsets这个topic中。

这里给了我们两个提示：

1、kafka自己管理客户端的消费进度，而不是依靠zk，这就是kafka官方文档说的kafka未来会抛弃zk的底气之一；

2、留意到这个kafka自己的topic是带下划线的，也就是，kafka担心我们自己建的topic如果带下划线的话会跟这些内部自用的topic冲突；

一.打包项目镜像:

利用Dockerfile 来打包项目的镜像

本次项目共依赖两个镜像(一个基础系统环境和一个项目镜像)

本次直接将Dockerfile写好后,用shell脚本build.sh启动打包:

然后切换到项目的目录下找到build.sh,运行即可打包项目镜像

若

报错:"failed to dial gRPC: cannot connect to the Docker daemon. Is 'docker daemon' running on this host?: dial unix /var/run/docker.sock: connect: permission denied

"

就用

出现以下说明打包成功,接下来可以开始部署:

https://jingyan.baidu.com/article/9113f81b49ed2f2b3214c7fa.html

注意:如果遇到只读权限不能修改时,将host文件复制一份到桌面,修改后在替换原来的host文件

在hosts文件末尾加上kafka服务器<!外网! 39. 0.25...>地址,修改后的格式如下:

1.1注意: 修改阿里云服务器的hosts 文件来配置 kafka的服务器地址:

在hosts 文件最后加入:

添加的 kafka-server 就是以下创建topic命令中的 kafka-server别名,

监听远程kafka:新建消费者:

远程创建topic的实例:

查看远程已创建的topc:

本地:

远程修改后的kafka topic:

2.通过git Bash 切换到kafka客户端的bin目录:

桌面打开 gitBash,切换到本地kafka软件目录:

这里一定要切换为windows

3.查看已经有的topic

--topic 指定topic名字

--replication-factor 指定副本数，因为我的是集群环境，这里副本数就为3

--partitions 指定分区数，这个参数需要根据broker数和数据量决定，正常情况下，每个broker上两个partition最好

注意:服务器部署时候一定要用内网172. .开头的,外部访问设为外网ip

不然会导致Kafka写入数据的时候报错 : TImeout

4.1本地docker创建topic：

4.2 本地windows 创建topic

进入本地软件路径KAFKA/BIN/WIONDOWS

创建topic

5.修改服务器的host:

一定要注意加sudo 不然会导致readonly 无法修改

在host 文件的末尾加上以下:

6.切换到工程部署的目录

7.清理redis,不然数据有残留:

7.1服务器上的redis挂载清除:

在 docker-compose.yml中注销这几行: 目的是每次启动不必记录上次没有执行完的数据.

这个是用来记录redis中假如上次指定的是1到100万块,没有执行完.下次接着执行没执行完的任务,测试时暂时关闭

7.2删除volume:

7.3 如果volume文件被占用时,先删除占用容器:

7.4 清除redis中的数据

进入redis容器中：

8.部署命令:

8.1开启docker可视化web上监控docker:

然后访问: http://39.100.48.41:9000

宿主机IP + 9000端口

8.2执行部署命令,启动服务:

9.部署时报错: yaml: line 46: did not find expected key

原因: docker-compose.yml文件中第46行 报错

解决:将所有数据对齐,不要有多余的空格.

---