爬虫架构|利用Kafka处理数据推送问题（1）

koflover

　　随着数据量越来越大，数据保存和数据存取的响应效率是有瓶颈的。
　　爬虫集群在向MySQL生产数据后，需要主动通知分发服务去消费数据，这样的通知机制是一种很低效的工作方式。
　　图1-1
　　基于这两个问题，我们选择使用Kafka来进行优化爬虫系统。
　　在这里给大家提供一个学习交流的平台，java架构师群： 558787436
　　一、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= (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>　　[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
　　作者：小怪聊职场
　　链接：https://www.jianshu.com/p/eb9dc3e31c06
　　來源：简书
　　简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。