前言

本文主要做了一件事，通过kafka横向扩展GeoEvent Server，构建GeoEvent+kafka集群部署。这件事有两个作用，其一，利用多台GeoEvent提高数据的吞吐量。其二，利用kafka集群的机制来提高GeoEvent Server软件的鲁棒性。

GeoEvent Server是用来实时接入的GIS数据的GIS实时服务器。而对于实时服务器来说，最重要的是系统的鲁棒性(Robustness)。什么是鲁棒性？鲁棒是Robust的音译，也就是健壮和强壮的意思。它是在异常和危险情况下系统生存的关键。比如说，计算机软件在输入错误、磁盘故障、网络过载或有意攻击情况下，能否不死机、不崩溃，就是该软件的鲁棒性。所谓“鲁棒性”，是指控制系统在一定（结构，大小）的参数摄动下，维持其它某些性能的特性。

自从Esri不推荐使用GeoEvent Server集群，而推荐采用GeoEvent Server单机部署后，GeoEvent Server的鲁棒性就成了用户关注的焦点。而现在，GeoEvent团队提供了一个方案，用来提高GeoEvent的鲁棒性以及横向扩展GeoEvent的能力。地址：http://www.arcgis.com/home/item.html?id=89048b1cfdda4a71a9b2b334fe8c8cc9

此教程旨在发布5篇文章，来分别叙述以下五个章节。本文为第二节。熟悉事件中心概念（Apache Kafka）。

目录

1 GeoEvent弹性扩展架构

这一部分的目的是建立企业和ArcGIS的多个节点上，将在整个教程的其余部分使用ArcGIS GeoEvent Server。

2 熟悉事件中心概念（Apache Kafka）

本节中的练习将帮助您安装和探索事件调度中心的特性，包括使用多个broker进行冗余和伸缩。

3 安装和配置一个分布式事件调度中心（Apache Kafka）

本节基于前一节，将指导您在三个节点上安装和配置分布式事件调度中心。在本教程中，您将在3台已配置的GeoEvent Server节点上安装事件调度中心。

4 配置GeoEvent Server启用分布式事件调度中心

本节的目的是在一个分布式的事件调度中心完成GeoEvent Server配置。具体来说，你将学习如何使用Kafka 连接器配置GeoEvent Server接收消费从Apache Kafka生产的信息。 自定义连接器可在ArcGIS GeoEvent Gallery找到。

5 探讨了GeoEvent Server利用分布式事件调度中心（Apache Kafka）实现鲁棒性

最后，在本节中，您将测试你的GeoEvent Server和kafka部署的鲁棒性，通过试验丢失brokers和丢失consumers，确保预期的消息仍然被接收并存储在系统中。

正文

2 熟悉事件中心概念（Apache Kafka）

       在本节中，您将探索事件中心的概念。在这种情况下，事件调度中心是接收消息并将它们分发给一个或多个输出的软件。理想的情况是，一个事件的中心也允许传入的消息可以并行来获得更好的性能。在本教程中，您将使用Apache卡夫卡作为事件调度中心。

       本节中的练习提供了通过命令行使用卡夫卡的基本介绍。有关更多信息，请参见官方的Apache Kafka文档。

       为了节省时间，建议这个练习你安装和实验kafka的是一个GeoEvent服务器节点的机器。

架构图

安装Apache Kafka和必要的组件

1.下载并安装jdk8 网址:http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

2.设置系统变量JAVA_HOME为C:\Program Files\Java\jdk1.8.0_111

   添加path：C:\Program Files\Java\jdk1.8.0_111\bin;

3.下载Apache Kafka 2.11-0.10.1.0 地址：https://www.apache.org/dyn/closer.cgi?path=/kafka/0.10.1.0/kafka_2.11-0.10.1.0.tgz

   解压到C:\并创建一个系统变量KAFKA_HOME=C:\kafka_2.11-0.10.1.0

4.启动Zookeeper

   ZooKeeper服务器软件，支持分布式协调和同步服务组。它也集中配置信息维护和命名。

   a.找到%KAFKA_HOME\config\zookeeper.properties 用记事本打开。

   b.更改clientPort一行。这个端口不能被机器上任何程序占用。更改clientPort=2181为          clientPort=2183

   c.更改日志文件存放目录为dataDir=C:\\tmp\\zookeeper

   d.保存并关闭zookeeper.properties

   e.打开命令行（管理员）

   f.输入如下命令来启动ZooKeeper：

    cd %KAFKA_HOME%

    title zookeeper

    bin\windows\zookeeper-server-start.bat config\zookeeper.properties

启动ZooKeeper

    可以看到出现了很多消息

Zookeeper已启动

   g.保持这个命令窗口开启，以方便等会停止zookeeper的操作。

   你已经在machine3成功的启动了Zookeeper。接下来的练习是指导你通过一个简单的kafka设置启动一个单broker的kafka。

使用单个broker配置和测试kafka

    kafka是一个分布式的消息系统。一个broker是kafka集群中的一个节点。

1.访问%KAFKA_HOME%\config\ 打开server.properties in a text editor。

2取消掉这句话的注释

    listeners=PLAINTEXT://:9092

3更新ZooKeeper端口

    zookeeper.connect=localhost:2183

4.增加一行如下内容，这将允许您删除话题。

    delete.topic.enable=true

5.日志路径，取决于windos版本

    log.dirs=C:\\tmp\\kafka-logs

6.新开启一个命令行运行kafka，使用下面的命令

    cd %KAFKA_HOME%

    title kafka0

    bin\windows\kafka-server-start.bat config\server.properties

启动kafka

    会看到各类消息出现，保持这个命令行窗口开启。

7.创建一个Kafka话题：

    kafka话题就是消息在kafka中的组织方式。消费者订阅话题，生产者将消息推送到话题上。一个话题本质上是流经系统的潜在的许多独立的数据流之一。

   a.打开一个命令行，创建一个分片数为1，复制因子为1的kafka话题。

    cd %KAFKA_HOME%

    title topics

    bin\windows\kafka-topics.bat --create --zookeeper localhost:2183 --replication-factor 1 --partitions 1 --topic test

创建kafka话题

       分片数是kafka话题被分割成的份数。你可以通过使用多个分区，使消息分割成多份同时进入多个kafka节点，并行这个话题，实现消息的高吞吐量。

       复制因子决定了一条消息被复制多少份。有一个以上的副本相当于有备份数据，有助于高可用性。在这个练习中，此kafka话题只有一个分区，因此每个消息只能有一个分区。

   c.使用以下命令可以查看Kafka的话题列表

    bin\windows\kafka-topics.bat --list --zookeeper localhost:2183

查看已存在的话题

   d.现在你的话题test已经创建完毕，最小化topics命令窗口

8.从Kafka话题消费接收消息。

   Kafka 消费者通过订阅话题来接收消息，消息的顺序按照出现在Kafka日志中的顺序。

   a.打开一个管理员权限的命令窗口。

   b.输入以下命令，在Kafka broker中创建一个消费者来接收“test”话题。

    cd %KAFKA_HOME%

    title consumer

    bin\windows\kafka-console-consumer.bat --zookeeper localhost:2183 --topic test --from-beginning

创建消费者

   输入完命令后，进入消费者模式。

   c.保持consumer命令行窗口可见。稍后会通过它查看接收消息的情况。

9.在Kafka话题上生产消息

       Kafka生产者推送消息到Kafka话题上。更具体的来说，一个生产者应用程序写入消息到话题分区的leader上，所以分区必须在写入发生前被选中。Kafka包含了消息分割和负载均衡的基本逻辑。无论如何，定制的生产者应用程序通过分析手头的工作流来实现这种逻辑。

   a.打开一个管理员命令行窗口。

   b.输入如下命令在Kafka broker上创建一个生产者，用来向test话题发送消息。注意指定broker正在运行的端口

    cd %KAFKA_HOME%

    title producer

    bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test

创建生产者

   输入完这些命令，你会进入生产者模式并且这个程序已经做好了准备去发送消息

   c.在producer命令窗口中输入一些文字，敲击回车键进行发送。

发送消息

   d.在消费者命令窗口，你可以看到相同的消息已经被此消费者接收。

消费消息

   在这个例子中，只有一个生产者和一个消费者，所以，这个消费者接收到了所有被生产的消息。

10.清空

   a.在消费者和生产者窗口都执行以下操作：

      i.敲击ctrl+c 结束这个进程。

      ii.在出现提示后，输入Y来终止批处理进程。

终止producer

      iii.两个进程都停止后，你可以关闭消费者的命令窗口，留下生产者的命令窗口用于接下来的练习。

   b.在topics命令窗口，输入以下命令来删除test话题。

      bin\windows\kafka-topics.bat --zookeeper localhost:2183 --delete --topic test

删除话题

   c.在kafka运行的kafka0命令窗口:

      i.输入Ctrl+c打断Kafka进程。

      ii.出现提示后，输入Y来终止这个命令行运行。

    这将确保想要删除的test话题被删除。

    留下kafka0命令窗口不要关闭，你会在下一个练习中用到它。

   d.不要对Zookeeper命令窗口做任何更改，ZooKeeper需要为了后面的练习继续运行。

使用3个broker配置和测试kafka

kafka可以在多个节点组成的集群中运行，以获得高可用性和可伸缩性。在这个练习中，你将通过在同一台计算机上创建3个broker来模拟多个节点。

1.将%KAFKA_HOME%\config\server.properties文件复制3份，分别命名为。

   server-0.properties

   server-1.properties

   server-2.properties

2.在每个文件中修改指定的broker id 分别为0,1和2.

   broker.id=0

3.在每一个文件中修改下面这行，在你的机器上指定不同的端口。

    tips：因为您将要在一台机器上运行3个brokers，每一个broker都需要指定一个不同的监听端口，如9092,9093和9094

    listeners=PLAINTEXT://:9092

4.在每一个文件中，修改日志路径为原始路径+broker编号。

如 log.dirs=C:\\tmp\\kafka-logs-0

5.在每个文件中，更新zookeeper端口为正在运行的这个zookeeper占用的端口。

   zookeeper.connect=localhost:2183

6.在文件底部，添加一行，通过它，允许删除话题。

# Delete enablement 

delete.topic.enable=true

7.启动3个kafka brokers

   a.打开3个管理员命令行，如果你在上一个练习中预留了kafka0命令行窗口没有关闭，你可以只打开2个命令行窗口。

   b.在每个命令行窗口中。输入以下命令启动kafka。指定3个不同的题目名称和不同的配置文件。

cd %KAFKA_HOME%

title kafka0

bin\windows\kafka-server-start.bat config\server-0.properties

启动3个kafka broker

你应该看到出现各种消息。你可以最小化命令窗口，但是保持他们是开启状态，这样你能轻易的停止kafka。

8.创建一个分布式的kafka话题

   a.在topics命令窗口，输入以下命令以创建一个具有3个分片，3个复制因子的kafka话题taxis。

     bin\windows\kafka-topics.bat --create --zookeeper localhost:2183 --replication-factor 3 --partitions 3 --topic taxis

创建话题

   b. 输入以下命令查看新话题taxis的描述

      bin\windows\kafka-topics.bat --describe --zookeeper localhost:2183 –-topic taxis

描述新话题

   在这个例子中：

   Broker 2 是 Partition 0的指挥者

   Broker 0 是 Partition 1的指挥者

   Broker 1 是 Partition 2的指挥者

   因为你有和副本数相同数量的broker。每个broker都维护每一个分片的一个副本。

   c.你可以最小化topics命令窗口。

9.在分布式kafka话题taxis上生产和消费消息

   a.以管理员身份打开3个命令窗口

   b.在三个命令窗口中输入以下命令分别创建3个taxis话题下的消费者。

     cd %KAFKA_HOME%

     title consumer0

     bin\windows\kafka-console-consumer.bat --zookeeper localhost:2183 --topic taxis

     cd %KAFKA_HOME%

     title consumer1

     bin\windows\kafka-console-consumer.bat --zookeeper localhost:2183 --topic taxis

     cd %KAFKA_HOME%

     title consumer2

     bin\windows\kafka-console-consumer.bat --zookeeper localhost:2183 --topic taxis

   c.在生产者命令窗口，输入以下命令来给taxis话题设置一个生产者。

      bin\windows\kafka-console-producer.bat --broker-list localhost:9092,localhost:9093,localhost:9094 --topic taxis

     为了确保可靠性，当消息写入一个话题时，生产者应用程序应该至少列出三个kafka broker，这样即使节点已经关闭，集群依然是可获取的。

   d.整理一下消费者命令窗口和生产者命令窗口，以便可以同时看到它们。

3个消费者1个生产者

   e.重头戏来了。在生产者命令窗口，输入消息并敲击回车键发送。例如：

     hello

     hello world

     it’s a good day

     Windows Server, 2012 R2, 64-bit

   f.你应该能看到相同的消息出现在了三个消费者命令窗口中。通过当前这种配置，每一个消费者将接收所有的消息。

每一个consumer接收所有的消息

g.在这四个命令窗口：

    i.使用Ctrl+c 杀掉进程

    ii.当出现提示时，输入Y来结束批处理任务

    iii.留下这些窗口不要关闭，用于下一步的练习。

10.在分布式的Kafka话题上生产和消费消息，通过配置多个分片，支持跨消费者的负载均衡

   a.在%KAFKA_HOME%\config文件夹，创建一个名叫common-consumer.config的文件。

   b.用文本编辑器编辑文件输入 group.id=taxis-consumer-group

   c.保存并关闭这个文件，这个文件定义了一个消费者群组，多个消费者都可以使用它。

   d.进入上一步留下来的命令窗口，输入以下命令，在taxis话题下创建三个消费者，共享同一个消费者群组。

      bin\windows\kafka-console-consumer.bat --zookeeper localhost:2183 --consumer.config config\common-consumer.config --topic taxis

共享同一个群组的消费者

   e.在生产者命令窗口，输入以下命令创建一个向taxis话题发送消息的生产者。

      bin\windows\kafka-console-producer.bat --broker-list localhost:9092,localhost:9093,localhost:9094 --topic taxis

  f.开始将文本信息输入到生产者命令窗口中，在消息之间单击回车键。

     使用这种方法，消费者只从分配给他们的话题分区中消费消息。卡夫卡的简单分区是基于消息散列的负载平衡，因此消息在消费者之间分布相对均匀。

消息分布均匀

对比9和10，可以清晰的看出Kafka消费者群组所起到的作用。

在这个练习中，你探索了在Kafka集群中配置多个消费者实现可靠性和扩展性（通过相同的消费者群组）。在下一个练习中，你将继续使用当前的Kafka配置来测试消费者的可靠性。

探索跨订阅者的kafka可靠性

在前面的练习中，你配置了三个代理卡夫卡集群、一个分布式主题和三个订阅者/消费者，它们都共享相同的公共消费组来消费消息。接下来，你将模拟订户间的失败以探索弹性。

1.排列好三个消费者命令窗口和一个生产者命令窗口以便你能同时看到它们。

   注意：在消费者窗口中依然有上一个练习中留下的消息。

上个练习中留下的消息

2.选择一个消费者窗口，Ctrl+c 然后输入y 结束这个批处理工作。

停掉一个消费者

   这个消费者进程被停掉后，意味着只有两个消费者共享相同的一个消费者群组了。

3.在生产者窗口中开始输入消息，敲击回车发送。当年发送消息，你会看到消息现在分布在剩下的两个消费者窗口中。

新消息分布在剩下的两个消费者

   当消费者共享一个相同的消费者群组，Kafka会根据消费者的数量变化动态地重新分配消息。

4.清除环境

   a.在正在运行的两个消费者命令窗口里，

     i.Ctrl+c打断进程。

     ii.出现提示后，输入Y终止这个批处理工作。

   你可以保留这些消费者窗口打开，你将在下一个练习中使用它们。

   b.在生产者命令窗口：

     i.点击Ctrl+c打断进程。

     ii.当出现提示后，输入Y终止这个批处理工作。

   保留producer窗口打开，下个练习中还要用到它们。

在这个练习中，您探索了kafka消费者的可靠性，观察到即使消费者的数量变化，消息仍会在当前的消费者之间被接收。在下一个练习中，您将探索跨broker的卡夫卡可靠性。

探索Kafka跨broker的可靠性

在前面的练习中，你探索了跨订阅者的Kafka可靠性。接下来，你将看到跨broker的可靠性。

1.排列3个消费者窗口和一个生产者窗口，以便你能同时看到它们。

3consumer1producer

2.在消费者命令窗口输入以下命令创建3个taxis话题下的消费者。

  注意：它们现在没有共享一个消费者群组。

  bin\windows\kafka-console-consumer.bat --zookeeper localhost:2183 --topic taxis

3.在生产者命令窗口输入以下命令设置一个taxis话题的生产者。

   bin\windows\kafka-console-producer.bat --broker-list localhost:9092,localhost:9093,localhost:9094 --topic taxis

4.在生产者命令窗口中，开始发送文本消息，观察到所有消息被每个消费者接收

每个消费者接收所有消息

5.现在，停止3个broker命令窗口中的任意一个。在前面的练习中，你设置他们的title为kafka0，kafka1，kafka2.

6.使用前一节中的方法终止掉你选择的那个kafka broker。

终止broker

7.现在，最小化那个命令窗口，查看你的消费者命令窗口。在每个消费者命令窗口中，你将会看到如下相似的消息。

减少一个broker

这表示broker中的一个已经消失，但这并不会打断或终止消费者进程的运行。

8.继续在生产者命令窗口中输入消息，你将会看到所有消息被每一个消费者接收到。

关闭broker依然接收消息

即使一个broker坏掉，所有的消费者依然可以接收本该他们接收到的消息。

9.清空环境。

   a.使用已有的经验来结束掉如下进程：

     i.Consumer 0-2

     ii.Producer

     iii.Kafka brokers 0-2

     iv.Zookeeper

10.当进程都结束后，关闭这些命令窗口。

在这个练习中，你使用单台计算机环境来确认了跨broker的kafka可靠性。在下一节中，你将在分布式机器上展开Kafka部署。

在上面的几个练习中，我们渐渐对kafka的消息机制有了概念。以及感受到了kafka自带的可靠性以及扩展性带来的便利。这让我们更加强烈的想让Kafka和GeoEvent结合，使GeoEvent也具有Kafka的这些能力。这一节真的非常重要同时也是非常基础、非常简单。希望这能引发你对Kafka消息调度中心的兴趣。