比特币挖矿——建立Kafka&ZooKeeper集群

1. 介绍

采用ZooKeeper+Kafka的方式建立集群，主要支持了消息传递和负载均衡，常见的一些文档，有的是概念较多，没有实际操作；有的只有命令，不知其原理。这里就结合集群矿池来说说它的具体应用场景，原理，以及具体实现。



图片.png

2. 举例

先举个例子，比如一个理发店：

一开始只有一个理发师，洗剪吹都由他一个人负责，那么只能一个客人做完全套（同一台机器上操作各个步骤），再给下一个客人服务。

之后, 又来了几个理发师，每个人负责接待一个客人（多台服务器，每台负责单个流程）。

再后来理发店又扩大，人多了（多台不同用途的机器），老板发现如果把工作细分为：洗(A)剪(B)吹(C)三部分，这三部分的工作量和难度都不同，可以安排专人专职，以节约人力成本。三个模块工作量不同，需要的人数也不同，如上图所示，洗2人(2台机器)，剪3人(3台机器)，吹2人(2台机器)。Data为库房(数据服务器)，每个人都可能去库房存取工具。

显然，A与B并不是一一对应的关系。洗发之后，应该分配剪发员执行下一步骤。让洗发员找每个剪发员查询状态显然不是明智的作法，而且，此时可能每个剪发员都不空闲，A还有接下来的工作，也不能一接等在那儿。

于是就引入了另一个角色：接待员broker，当某个A的工作做完全后，则向broker的topic发一个消息，告诉它本层已完成，需要下一层处理，此时A是发送者producer，中介是broker（它也位于一台机器上，这里没用方框画出来），如果某个B空闲时，也会连接到topic，等待下一个工作（如上图所示），此时B是接受者customer。B和C的通讯也是一样，不过它们间通讯时B是producer，C是customer。

Broker是Kafka的一个实例，broker按功能不同又可建立多个topic，topic里又可包含多个partition，每个partition都可保证先入先出，但partition之间不保证顺序。多个broker又可以做成Kafka群组。综上，Kafka是传送消息的工具，或者说是一套机制，它实现了消息创建，存取等工作。



图片.png

在producer，customer，broker的背后又有着ZooKeeper的支持，它就好像理发店的排班员，在这里主要负责负载均衡，将工作平均分配给customer，避免产生闲的闲死忙的忙死这种问题。ZooKeeper也可以有多个，即使某一个死掉，整个系统仍可正常运转。

从此处可见Kafka和Zookeeper上跑的不是执行具体工作的程序（洗剪吹），它们的角色是提供通讯和负载均衡的协调者。具体洗剪吹运行在哪个服务器上是另外的工作．

3. 配置过程

1) 配置ZooKeeper

在一台或几台机器上配置和运行ZooKeeper服务。

2) 配置Kafka

配置Kafka的 broker

创建topic及partition。

3) 修改程序

在程序中分别加入对producer和consumer的调用，以传递信息。

4) 说明

通过以上步骤，程序就可以在不同的机器通过简单地调用函数传递消息了。上述的ZooKeeper，Kafka，程序可以运行在同一台机器上，可以分别在运行在不同机器上。

4. ZooKeeper

1) 介绍

ZooKeeper是一个快速、高可靠、容错、分布式的协调服务。

2) 配置

$ cp conf/zoo_sample.cfg conf/zoo.cfg
$ vi conf/zoo.cfg #修改配置文件

配置文件中server*是zookeeper服务器之间交互用的，它使得一台服务器死机时，另外一台，能快速转换为主zookeeper服务器

dataDir是数据目录，clientPort为ZooKepper端口号，一般不用改

$ ./startserver.sh start # 启动zooKeeper服务

3) 调试工具

$ bin/zkCli.sh -server 127.0.0.1:2181 #client端连上看一下
$ jps #查看进程, 其中QuorumPeerMain是zooKeeper的主进程
$ netstat -nap|grep 2181 # 查看配置文件中设置的端口是否打开

正常显示为：

tcp  0   0   0.0.0.0:2181   0.0.0.0:*   LISTEN   10737/java

5. Kafka

1) 介绍

Kafka是一个分布式的流数据处理平台。

2) 配置文件

$ vi config/server.properties
这是默认的配置文件，以下几项比较重要
broker.id=1 # broker的标识，具有唯一性
port=9092 # kafka服务的端口号
log.dirs=/tmp/kafka-logs # 存储log的目录，交互的信息就存在该目录下
zookeeper.connect=127.0.0.1:2181 # zookeeper服务器的IP和端口,多个用逗号分开
host.name=127.0.0.1 # 运行本程序机器的IP地址，在btcpool的cfg中指定
delete.topic.enable=true #建议加这句，否则删topic时候特别麻烦
message.max.bytes=20000000 # 支持大数据msg，否则传输时将丢弃大数据块
replica.fetch.max.bytes=30000000 # 同上

3) 启动服务

$ bin/kafka-server-start.sh config/server.properties

4) 创建供btcpool使用的topic

需要先启后台服务，再创建topic，我这里使用了一个本机的zookeeper，如果有多个，请使用逗号分隔。

$ ./bin/kafka-topics.sh --create --topic test --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1

5) 相关工具

i. 查看topic

$ ./bin/kafka-topics.sh --describe --zookeeper 127.0.0.1:2181 #查看topic

ii. 查看kafka服务

$ netstat -nap|grep 9092

正常运行时返回

tcp  0   0 127.0.0.1:9092   0.0.0.0:*   LISTEN   24961/java

6) 删除topic

先删除kafka存储目录（server.properties文件log.dirs配置，默认为"/tmp/kafka-logs"）相关topic目录。注意partition的数据都存在这个目录中。

./bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --topic test -delete

如果在server.properties中未指定delete.topic.enable=true，则只会做一个删除标记，不会真正删除，删除需要在zookeeper服务器上，通过zkCli.sh连进去删除。

7) 常出现的错误

i. kafka-server-start.sh报错找不到主机，然后直接退出

解决方法，在配置文件中设置主机

host.name=主机名

主机名通过命令hostname查看

ii. kafka-server-start.sh虽不退出，但不停报错java.lang.NoClassDefFoundError: Could not initialize class kafka.network.RequestChannel$

这也是主机名相关的问题，在/etc/hosts的最后填加一句

127.0.0.1 主机名

主机名通过命令hostname查看

8) 测试kafka

正常情况下，producer上发的，consumer都应该能收到

i. 创建

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

ii. 接收

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

iii. 发送

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

如果topic没建立，producer会自动建立top，但参数可能与你想要的不同．

6. C程序调用kafka发消息

一般使用librdkafka库，下载库源码编译，其中example目录中的程序也可以用于测试。

7. 参考

1) Apache Kafka 分布式消息队列中间件安装与配置

http://blog.csdn.net/wangjia184/article/details/37921183

2) Kafka&Zookeeper原理与应用场景介绍

https://wenku.baidu.com/view/a47389116ad97f192279168884868762caaebbd4.html