下载 1.0.0版本并解压缩。java
> tar -xzf kafka_2.11-1.0.0.tgz
> cd kafka_2.11-1.0.0
Kafka使用ZooKeeper,因此若是你尚未ZooKeeper服务器,你须要先启动一个ZooKeeper服务器。您可使用与kafka一块儿打包的便捷脚原本获取快速而简单的单节点ZooKeeper实例。apache
> bin/zookeeper-server-start.sh config/zookeeper.properties [2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig) ...
如今启动Kafka服务器:bootstrap
> bin/kafka-server-start.sh config/server.properties [2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties) [2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties) ...
咱们用一个分区和一个副本建立一个名为“test”的主题:bash
> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
咱们如今能够看到这个主题,若是咱们运行列表主题命令:服务器
> bin/kafka-topics.sh --list --zookeeper localhost:2181
test
或者,您也能够将代理配置为在发布不存在的主题时自动建立主题,而不是手动建立主题。socket
Kafka带有一个命令行客户端,它将从文件或标准输入中获取输入,并将其做为消息发送到Kafka集群。默认状况下,每行将做为单独的消息发送。分布式
运行生产者,而后在控制台输入一些消息发送到服务器。微服务
> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
This is a message
This is another message
Kafka也有一个命令行消费者,将消息转储到标准输出。工具
> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
This is a message
This is another message
若是您将上述每一个命令都运行在不一样的终端中,则如今应该能够将消息键入生产者终端,并将其显示在消费者终端中。测试
全部的命令行工具都有其余选项。不带任何参数运行该命令将显示使用信息更详细地记录它们。
到目前为止,咱们一直在与一个broker竞争,但这并很差玩。对于kafka来讲,一个broker只是一个规模一个的集群,因此除了开始一些borker实例以外没有太大的变化。可是为了获得它的感受,让咱们把咱们的集群扩展到三个节点(仍然都在咱们的本地机器上)。
首先,咱们为每一个代理建立一个配置文件(在Windows上使用copy
命令代替):
> cp config/server.properties config/server-1.properties
> cp config/server.properties config/server-2.properties
如今编辑这些新文件并设置如下属性:
config/server-1.properties: broker.id=1 listeners=PLAINTEXT://:9093 log.dir=/tmp/kafka-logs-1 config/server-2.properties: broker.id=2 listeners=PLAINTEXT://:9094 log.dir=/tmp/kafka-logs-2
该broker.id
属性是群集中每一个节点的惟一且永久的名称。咱们必须重写端口和日志目录,由于咱们在同一台机器上运行这些端口和日志目录,咱们但愿让全部的代理都试图在同一个端口注册或覆盖彼此的数据。
咱们已经有Zookeeper和咱们的单节点了,因此咱们只须要启动两个新的节点:
> bin
/kafka-server-start
.sh config
/server-1
.properties &
...
> bin
/kafka-server-start
.sh config
/server-2
.properties &
...
如今建立一个ReplicationFactor为3的新主题:
> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
好吧,如今咱们有一个集群,咱们怎么知道哪一个broker在作什么?要查看运行“描述主题”命令:
> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs: Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
这里是对输出的解释。第一行给出了全部分区的摘要,每一个附加行给出了关于一个分区的信息。因为咱们只有一个分区,因此只有一行。
请注意,在个人示例中,节点1是该主题的惟一分区的领导者。
咱们能够在咱们建立的原始主题上运行相同的命令来查看它的位置:
> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test Topic:test PartitionCount:1 ReplicationFactor:1 Configs: Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0
因此这里并不奇怪,原来的主题没有副本,并且在服务器0上,这是咱们建立集群时惟一的服务器。
让咱们发表一些信息给咱们的新主题:
> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic ... my test message 1 my test message 2 ^C
如今让咱们消费这些消息:
> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic ... my test message 1 my test message 2 ^C
如今咱们来测试一下容错。broker1是做为领导者,因此让咱们杀了它:
> ps aux | grep server-1.properties 7564 ttys002 0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.8/Home/bin/java... > kill -9 7564
在Windows上使用:
> wmic process where "caption = 'java.exe' and commandline like '%server-1.properties%'" get processid
ProcessId
6016
> taskkill /pid 6016 /f
领导已经切换到其中一个从属节点,而且节点1再也不处于同步副本集合中:
> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs: Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 1,2,0 Isr: 2,0
可是,即便原先写入的领导者失败,这些消息仍然可用于消费:
> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic ... my test message 1 my test message 2 ^C
从控制台写入数据并将其写回控制台是一个方便的起点,但您可能想要使用其余来源的数据或将数据从Kafka导出到其余系统。对于许多系统,您可使用Kafka Connect来导入或导出数据,而没必要编写自定义集成代码。
Kafka Connect是Kafka包含的一个工具,能够将数据导入和导出到Kafka。它是一个可扩展的工具,运行 链接器,实现与外部系统交互的自定义逻辑。在这个快速入门中,咱们将看到如何使用简单的链接器运行Kafka Connect,这些链接器将数据从文件导入到Kafka主题,并将数据从Kafka主题导出到文件。
首先,咱们将经过建立一些种子数据开始测试:
> echo -e "foo\nbar" > test.txt
或在Windows上:
> echo foo> test.txt
> echo bar>> test.txt
接下来,咱们将启动两个以独立模式运行的链接器,这意味着它们将在单个本地专用进程中运行。咱们提供三个配置文件做为参数。首先是Kafka Connect过程的配置,包含常见的配置,例如要链接的Kafka代理以及数据的序列化格式。其他的配置文件都指定一个要建立的链接器。这些文件包括惟一的链接器名称,要实例化的链接器类以及链接器所需的任何其余配置。
> bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties
Kafka附带的这些示例配置文件使用您以前启动的默认本地群集配置,并建立两个链接器:第一个是源链接器,用于从输入文件中读取行,并将每一个链接生成为Kafka主题,第二个为链接器链接器它从Kafka主题读取消息,并在输出文件中产生每行消息。
在启动过程当中,您会看到一些日志消息,其中一些指示链接器正在实例化。Kafka Connect进程启动后,源链接器应该开始读取test.txt
主题connect-test
,并将其生成主题,而且接收器链接器应该开始读取主题中的消息connect-test
并将其写入文件test.sink.txt
。咱们能够经过检查输出文件的内容来验证经过整个管道传输的数据:
> more test.sink.txt
foo
bar
请注意,数据存储在Kafka主题中connect-test
,所以咱们也能够运行控制台使用者来查看主题中的数据(或使用自定义使用者代码来处理它):
> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning {"schema":{"type":"string","optional":false},"payload":"foo"} {"schema":{"type":"string","optional":false},"payload":"bar"} ...
链接器继续处理数据,因此咱们能够将数据添加到文件中,并看到它在管道中移动:
> echo Another line>> test.txt
您应该看到该行出如今控制台使用者输出和接收器文件中。
Kafka Streams是用于构建关键任务实时应用程序和微服务的客户端库,输入和/或输出数据存储在Kafka集群中。Kafka Streams结合了在客户端编写和部署标准Java和Scala应用程序的简单性以及Kafka服务器端集群技术的优点,使这些应用程序具备高度可伸缩性,弹性,容错性,分布式等特性。本快速入门示例将演示如何运行在此库中编码的流式应用程序。