如下图1-1所示，我们之前爬虫集群在采集完数据之后是直接插入到MySQL数据库中，分发服务再消费MySQL里面的数据。这样的设计会有两个主要的问题：
- 随着数据量越来越大，数据保存和数据存取的响应效率是有瓶颈的。
- 爬虫集群在向MySQL生产数据后，需要主动通知分发服务去消费数据，这样的通知机制是一种很低效的工作方式。
图1-1
基于这两个问题，我们选择使用Kafka来进行优化爬虫系统。
一、Kafka介绍
Kafka是一个分布式的、可分区的、可复制的消息系统。它提供了普通消息系统的功能，但具有自己独特的设计，它的独特之处主要体现如下几点：
1、Kafka将消息以topic为单位进行归纳。
2、将向Kafka topic发布消息的程序成为producers。
3、将预订topics并消费消息的程序成为consumer。
4、Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker。
5、producers通过网络将消息发送到Kafka集群，集群向消费者提供消息，如下图1-2所示：
图1-2
客户端和服务端通过TCP协议通信。Kafka提供了Java客户端，并且对多种语言都提供了支持。
1.1、Topics 和Logs
先来看一下Kafka提供的一个抽象概念：topic.
一个topic是对一组消息的归纳。对每个topic，Kafka 对它的日志进行了分区，如下图1-3所示：
图1-3
每个分区都由一系列有序的、不可变的消息组成，这些消息被连续的追加到分区中。分区中的每个消息都有一个连续的序列号叫做offset，用来在分区中唯一的标识这个消息。
在一个可配置的时间段内，Kafka集群保留所有发布的消息，不管这些消息有没有被消费。比如，如果消息的保存策略被设置为2天，那么在一个消息被发布的两天时间内，它都是可以被消费的。之后它将被丢弃以释放空间。Kafka的性能是和数据量无关的常量级的，所以保留太多的数据并不是问题。
实际上每个consumer唯一需要维护的数据是消息在日志中的位置，也就是offset。这个offset由consumer来维护：一般情况下随着consumer不断的读取消息，这offset的值不断增加，但其实consumer可以以任意的顺序读取消息，比如它可以将offset设置成为一个旧的值来重读之前的消息。
以上特点的结合，使Kafka consumers非常的轻量级：它们可以在不对集群和其他consumer造成影响的情况下读取消息。你可以使用命令行来“tail”消息而不会对其他正在消费消息的consumer造成影响。
将日志分区可以达到以下目的：首先这使得每个日志的数量不会太大，可以在单个服务上保存。另外每个分区可以单独发布和消费，为并发操作topic提供了一种可能。
1.2、分布式
每个分区在Kafka集群的若干服务中都有副本，这样这些持有副本的服务可以共同处理数据和请求，副本数量是可以配置的。副本使Kafka具备了容错能力。
每个分区都由一个服务器作为“leader”，零或若干服务器作为“followers”，leader负责处理消息的读和写，followers则去复制leader。如果leader down了，followers中的一台则会自动成为leader。集群中的每个服务都会同时扮演两个角色：作为它所持有的一部分分区的leader，同时作为其他分区的followers，这样集群就会据有较好的负载均衡。
1.3、Producers
Producer将消息发布到它指定的topic中，并负责决定发布到哪个分区。通常简单的由负载均衡机制随机选择分区，但也可以通过特定的分区函数选择分区。使用的更多的是第二种。
1.4、Consumers
发布消息通常有两种模式：队列模式（queuing）和发布-订阅模式(publish-subscribe)。队列模式中，consumers可以同时从服务端读取消息，每个消息只被其中一个consumer读到；发布-订阅模式中消息被广播到所有的consumer中。
Consumers可以加入一个consumer组，共同竞争一个topic，topic中的消息将被分发到组中的一个成员中。同一组中的consumer可以在不同的程序中，也可以在不同的机器上。如果所有的consumer都在一个组中，这就成为了传统的队列模式，在各consumer中实现负载均衡。如果所有的consumer都不在不同的组中，这就成为了发布-订阅模式，所有的消息都被分发到所有的consumer中。更常见的是，每个topic都有若干数量的consumer组，每个组都是一个逻辑上的“订阅者”，为了容错和更好的稳定性，每个组由若干consumer组成。这其实就是一个发布-订阅模式，只不过订阅者是个组而不是单个consumer。如下图1-4所示：
图1-4
由两个机器组成的集群拥有4个分区 (P0-P3) 。2个Consumer组，A组有两个Consumer，B组有4个Consumer。
相比传统的消息系统，Kafka可以很好的保证有序性。
传统的队列在服务器上保存有序的消息，如果多个consumers同时从这个服务器消费消息，服务器就会以消息存储的顺序向consumer分发消息。虽然服务器按顺序发布消息，但是消息是被异步的分发到各consumer上，所以当消息到达时可能已经失去了原来的顺序，这意味着并发消费将导致顺序错乱。为了避免故障，这样的消息系统通常使用“专用consumer”的概念，其实就是只允许一个消费者消费消息，当然这就意味着失去了并发性。
在这方面Kafka做的更好，通过分区的概念，Kafka可以在多个consumer组并发的情况下提供较好的有序性和负载均衡。将每个分区分只分发给一个consumer组，这样一个分区就只被这个组的一个consumer消费，就可以顺序的消费这个分区的消息。因为有多个分区，依然可以在多个consumer组之间进行负载均衡。注意consumer组的数量不能多于分区的数量，也就是有多少分区就允许多少并发消费。
Kafka只能保证一个分区之内消息的有序性，在不同的分区之间是不可以的，这已经可以满足大部分应用的需求。如果需要topic中所有消息的有序性，那就只能让这个topic只有一个分区，当然也就只有一个consumer组消费它。
二、环境搭建
2.1、Kafka安装（Mac下）
brew install kafka
安装会依赖zookeeper。
看到如下结果，表示安装成功了。
Updating Homebrew...
==> Auto-updated Homebrew!
Updated 2 taps (homebrew/core, pivotal/tap).
==> New Formulae
bluepill gobuster prest
buildifier grunt sha1dc
crowdin gtef shellshare
dhall-json libcds spigot
dmtx-utils librealsense teleport
docker-credential-helper libtensorflow uniutils
dvd-vr monitoring-plugins urh
gandi.cli mysql-utilities woboq_codebrowser
go@1.7 pqiv
==> Updated Formulae
kops wakatime-cli
kotlin wesnoth
kubernetes-cli winetricks
kubernetes-helm wireguard-tools
ldc wireshark
leveldb x265
lftp xmake
libarchive xonsh
libav yarn
libbluray yle-dl
libchamplain youtube-dl
libcouchbase zimg
libdvbpsi znc
libepoxy zsh-history-substring-search
libfabric
==> Renamed Formulae
bash-completion2 -> bash-completion@2
bazel02 -> bazel@0.2
bigdata -> blazegraph
bison27 -> bison@2.7
boost-python159 -> boost-python@1.59
cassandra21 -> cassandra@2.1
cassandra22 -> cassandra@2.2
clang-format38 -> clang-format@3.8
docker111 -> docker@1.11
docker171 -> docker@1.71
erlang-r18 -> erlang@18
ffmpeg28 -> ffmpeg@2.8
freetds091 -> freetds@0.91
giflib5 -> giflib@5
glfw2 -> glfw@2
gnupg21 -> gnupg@2.1
gnuplot4 -> gnuplot@4
go14 -> go@1.4
go15 -> go@1.5
go16 -> go@1.6
gradle214 -> gradle@2.14
grails25 -> grails@2.5
gsl1 -> gsl@1
gst-ffmpeg010 -> gst-ffmpeg@0.10
gst-plugins-bad010 -> gst-plugins-bad@0.10
gst-plugins-base010 -> gst-plugins-base@0.10
gst-plugins-good010 -> gst-plugins-good@0.10
gst-plugins-ugly010 -> gst-plugins-ugly@0.10
gstreamer010 -> gstreamer@0.10
mariadb100 -> mariadb@10.0
postgresql94 -> postgresql@9.4
postgresql95 -> postgresql@9.5
rebar3 -> rebar@3
recipes -> gnome-recipes
redis26 -> redis@2.6
redis28 -> redis@2.8
ruby187 -> ruby@1.8
ruby193 -> ruby@1.9
ruby20 -> ruby@2.0
ruby21 -> ruby@2.1
ruby22 -> ruby@2.2
ruby23 -> ruby@2.3
==> Deleted Formulae
ctorrent libgroove s3sync silc-client
ee node@5 scsh xstow
==> Installing dependencies for kafka: zookeeper
==> Installing kafka dependency: zookeeper
==> Downloading https://homebrew.bintray.com/bottles/zookeeper-3.4.9.sierra.bott
######################################################################## 100.0%
==> Pouring zookeeper-3.4.9.sierra.bottle.tar.gz
==> Caveats
To have launchd start zookeeper now and restart at login:
brew services start zookeeper
Or, if you don't want/need a background service you can just run:
zkServer start
==> Summary
�� /usr/local/Cellar/zookeeper/3.4.9: 238 files, 18.2MB
==> Installing kafka
==> Downloading https://homebrew.bintray.com/bottles/kafka-0.10.2.0.sierra.bottl
######################################################################## 100.0%
==> Pouring kafka-0.10.2.0.sierra.bottle.tar.gz
==> Caveats
To have launchd start kafka now and restart at login:
brew services start kafka
Or, if you don't want/need a background service you can just run:
zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties & kafka-server-start /usr/local/etc/kafka/server.properties
==> Summary
/usr/local/Cellar/kafka/1.0.0: 132 files, 37.2MB
从上面的最后结果得知安装目录为：/usr/local/Cellar/kafka/1.0.0
2.2、安装的配置文件位置
/usr/local/etc/kafka/server.properties
/usr/local/etc/kafka/zookeeper.properties
2.3、启动zookeeper
cd /usr/local/Cellar/kafka/1.0.0
./bin/zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties &
执行之后：
/usr/local/etc/kafka/zookeeper.properties &
[1] 4436
lidongdeMacBook-Pro:0.10.2.0 lidong$ [2017-03-15 14:11:38,682] INFO Reading configuration from: /usr/local/etc/kafka/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2017-03-15 14:11:38,685] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
[2017-03-15 14:11:38,685] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
[2017-03-15 14:11:38,685] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
[2017-03-15 14:11:38,685] WARN Either no config or no quorum defined in config, running in standalone mode (org.apache.zookeeper.server.quorum.QuorumPeerMain)
[2017-03-15 14:11:38,708] INFO Reading configuration from: /usr/local/etc/kafka/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2017-03-15 14:11:38,708] INFO Starting server (org.apache.zookeeper.server.ZooKeeperServerMain)
[2017-03-15 14:11:38,728] INFO Server environment:zookeeper.version=3.4.9-1757313, built on 08/23/2016 06:50 GMT (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,728] INFO Server environment:host.name=192.168.1.130 (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,728] INFO Server environment:java.version=1.8.0_92 (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,728] INFO Server environment:java.vendor=Oracle Corporation (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,728] INFO Server environment:java.home=/Library/Java/JavaVirtualMachines/jdk1.8.0_92.jdk/Contents/Home/jre (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,728] INFO Server environment:java.class.path=:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/aopalliance-repackaged-2.5.0-b05.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/argparse4j-0.7.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/connect-api-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/connect-file-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/connect-json-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/connect-runtime-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/connect-transforms-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/guava-18.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/hk2-api-2.5.0-b05.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/hk2-locator-2.5.0-b05.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/hk2-utils-2.5.0-b05.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jackson-annotations-2.8.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jackson-annotations-2.8.5.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jackson-core-2.8.5.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jackson-databind-2.8.5.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jackson-jaxrs-base-2.8.5.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jackson-jaxrs-json-provider-2.8.5.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jackson-module-jaxb-annotations-2.8.5.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/javassist-3.20.0-GA.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/javax.annotation-api-1.2.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/javax.inject-1.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/javax.inject-2.5.0-b05.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/javax.servlet-api-3.1.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/javax.ws.rs-api-2.0.1.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jersey-client-2.24.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jersey-common-2.24.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jersey-container-servlet-2.24.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jersey-container-servlet-core-2.24.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jersey-guava-2.24.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jersey-media-jaxb-2.24.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jersey-server-2.24.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jetty-continuation-9.2.15.v20160210.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jetty-http-9.2.15.v20160210.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jetty-io-9.2.15.v20160210.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jetty-security-9.2.15.v20160210.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jetty-server-9.2.15.v20160210.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jetty-servlet-9.2.15.v20160210.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jetty-servlets-9.2.15.v20160210.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jetty-util-9.2.15.v20160210.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/jopt-simple-5.0.3.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/kafka-clients-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/kafka-log4j-appender-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/kafka-streams-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/kafka-streams-examples-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/kafka-tools-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/kafka_2.11-0.10.2.0-sources.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/kafka_2.11-0.10.2.0-test-sources.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/kafka_2.11-0.10.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/log4j-1.2.17.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/lz4-1.3.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/metrics-core-2.2.0.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/osgi-resource-locator-1.0.1.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/reflections-0.9.10.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/rocksdbjni-5.0.1.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/scala-library-2.11.8.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/scala-parser-combinators_2.11-1.0.4.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/slf4j-api-1.7.21.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/slf4j-log4j12-1.7.21.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/snappy-java-1.1.2.6.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/validation-api-1.1.0.Final.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/zkclient-0.10.jar:/usr/local/Cellar/kafka/0.10.2.0/libexec/bin/../libs/zookeeper-3.4.9.jar (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:java.library.path=/Users/lidong/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:. (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:java.io.tmpdir=/var/folders/vl/zhy0vf8141vb8y5f6yx0dgrr0000gn/T/ (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:java.compiler=<NA> (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:os.name=Mac OS X (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:os.arch=x86_64 (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:os.version=10.12.3 (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:user.name=lidong (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:user.home=/Users/lidong (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,729] INFO Server environment:user.dir=/usr/local/Cellar/kafka/0.10.2.0 (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,743] INFO tickTime set to 3000 (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,743] INFO minSessionTimeout set to -1 (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,743] INFO maxSessionTimeout set to -1 (org.apache.zookeeper.server.ZooKeeperServer)
[2017-03-15 14:11:38,769] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)
2.4、启动Kafka服务
cd /usr/local/Cellar/kafka/1.0.0
./bin/kafka-server-start /usr/local/etc/kafka/server.properties &
执行结果如下：
[1] 4751
$ [2017-03-15 14:13:36,327] INFO KafkaConfig values:
advertised.host.name = null
advertised.listeners = null
advertised.port = null
authorizer.class.name =
auto.create.topics.enable = true
auto.leader.rebalance.enable = true
background.threads = 10
broker.id = 0
broker.id.generation.enable = true
broker.rack = null
......
......
......
[2017-03-15 14:13:37,145] INFO [ExpirationReaper-0], Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2017-03-15 14:13:37,158] INFO [GroupCoordinator 0]: Starting up. (kafka.coordinator.GroupCoordinator)
[2017-03-15 14:13:37,159] INFO [GroupCoordinator 0]: Startup complete. (kafka.coordinator.GroupCoordinator)
[2017-03-15 14:13:37,161] INFO [Group Metadata Manager on Broker 0]: Removed 0 expired offsets in 2 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2017-03-15 14:13:37,187] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader$)
[2017-03-15 14:13:37,218] INFO Creating /brokers/ids/0 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
[2017-03-15 14:13:37,225] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
[2017-03-15 14:13:37,227] INFO Registered broker 0 at path /brokers/ids/0 with addresses: EndPoint(192.168.1.130,9092,ListenerName(PLAINTEXT),PLAINTEXT) (kafka.utils.ZkUtils)
[2017-03-15 14:13:37,229] WARN No meta.properties file under dir /usr/local/var/lib/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
[2017-03-15 14:13:37,236] INFO New leader is 0 (kafka.server.ZookeeperLeaderElector$LeaderChangeListener)
[2017-03-15 14:13:37,259] INFO Kafka version : 0.10.2.0 (org.apache.kafka.common.utils.AppInfoParser)
[2017-03-15 14:13:37,259] INFO Kafka commitId : 576d93a8dc0cf421 (org.apache.kafka.common.utils.AppInfoParser)
[2017-03-15 14:13:37,260] INFO [Kafka Server 0], started (kafka.server.KafkaServer)
2.5、创建topic
使用单个分区和只有一个副本创建一个名为“test”的主题：
cd /usr/local/Cellar/kafka/1.0.0
./bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
执行结果如下：
Created topic "test".
2.6、查看创建的topic
./bin/kafka-topics --list --zookeeper localhost:2181
2.7、发送一些消息
Kafka提供了一个命令行客户端，它将从文件或标准输入接收输入，并将其作为消息发送到Kafka集群。默认情况下，每行都将作为单独的消息发送。
运行生产者，然后在控制台中键入一些消息发送到服务器。
cd /usr/local/Cellar/kafka/1.0.0
./bin/kafka-console-producer --broker-list localhost:9092 --topic test
第一条消息
第二条消息
2.8、消费消息
Kafka还有一个命令行消费者，将消息转储到标准输出。
cd /usr/local/Cellar/kafka/1.0.0
./bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic test --from-beginning
# 输出
第一条消息
第二条消息
发送消息和消费消息命令运行在不同的终端，当输入消息到生产者终端，可以看到他们立马出现在消费者终端。
三、使用阿里云的Kafka（此处无意为阿里云打广告）
你可以自己购买服务器搭建一个Kafka（如上方式搭建），当然也可以直接使用阿里云的Kafka服务。
使用阿里云的Kafka服务可以直接参考它的快速入门文档，简单明了。
本章节主要描述从开通消息队列 Kafka 服务、创建消息队列 Kafka 资源，到使用消息队列 Kafka 进行消息收发的完整流程，旨在以最简单明了的方式引导您快速上手消息队列 Kafka，为进一步使用和熟悉消息队列 Kafka 的功能提供入门。
消息队列 Kafka 快速接入步骤如下：
步骤一：开通服务
步骤二：创建资源
步骤三：准备配置
步骤四：发布消息
步骤五：订阅消息
这里注意说一下我在操作是遇到的两个问题，下图1-5：
- 创建Topic时要选择使用公网的方式，其他是生产环境，外网无法访问。
- 接入点在下图标示为2的地方，如下在配置kafka.properties文件时，我死活找不到接入点到底在哪。
## 接入点，通过控制台获取
## 您在控制台获取的接入点
bootstrap.servers=kafka-cn-internet.aliyun.com:8080
## Topic，通过控制台创建
## 您在控制台创建的Topic
topic=alikafka-topic-demo
## Consumer Grouo，通过控制台创建
## 您在控制台创建的 Consumer Group
group.id=CID-consumer-group-demo
## ssl 根证书的路径，demo中有，请拷贝到自己的某个目录下，不能被打包到jar中
## 这里假设您的目录为/home/admin，请记得修改为自己的实际目录
ssl.truststore.location=/home/admin/kafka.client.truststore.jks
## sasl路径，demo中有，请拷贝到自己的某个目录下，不能被打包到jar中
## 这里假设您的目录为/home/admin，请记得修改为自己的实际目录
java.security.auth.login.config=/home/admin/kafka_client_jaas.conf
图1-5
网友评论