经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置:技术经验 » 大数据/云/AI » Apache Kafka » 查看文章
Kafka学习(一)-------- Quickstart
来源:cnblogs  作者:独孤风  时间:2019/7/9 9:18:31  对本文有异议

参考官网:http://kafka.apache.org/quickstart

一、下载Kafka

官网下载地址 http://kafka.apache.org/downloads

截至2019年7月8日 最新版本为 2.3.0 2.12为编译的scala版本 2.3.0为kafka版本

  • Scala 2.12 ?- kafka_2.12-2.3.0.tgz (asc, sha512)
    解压
    > tar -xzf kafka_2.12-2.3.0.tgz
    > cd kafka_2.12-2.3.0

二、启动服务

要先启动zookeeper kafka内置了一个 也可以不用

  1. > bin/zookeeper-server-start.sh config/zookeeper.properties
  2. [2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
  3. ...
  4. > bin/kafka-server-start.sh config/server.properties
  5. [2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties)
  6. [2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties)
  7. ...

三、创建topic

  1. replication-factor1 partitions1
  2. > bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
  3. 查看topic
  4. > bin/kafka-topics.sh --list --bootstrap-server localhost:9092
  5. test

也可以不创建topic 设置自动创建 当publish的时候

四、发送消息

用command line client 进行测试 一行就是一条消息

  1. > bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
  2. This is a message
  3. This is another message

五、消费者

command line consumer 可以接收消息

  1. > bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
  2. This is a message
  3. This is another message

六、设置多broker集群

单broker没有意思 我们可以设置三个broker

首先为每个broker 复制配置文件

  1. > cp config/server.properties config/server-1.properties
  2. > cp config/server.properties config/server-2.properties

然后编辑

  1. config/server-1.properties:
  2. broker.id=1
  3. listeners=PLAINTEXT://:9093
  4. log.dirs=/tmp/kafka-logs-1
  5. config/server-2.properties:
  6. broker.id=2
  7. listeners=PLAINTEXT://:9094
  8. log.dirs=/tmp/kafka-logs-2

broker.id是唯一的 cluster中每一个node的名字 我们在same machine上 所有要设置listeners和log.dirs 以防冲突

建一个topic 一个partitions 三个replication-factor

  1. > bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 1 --topic my-replicated-topic
  2. describe看看都是什么情况
  3. > bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic my-replicated-topic
  4. Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
  5. Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
  • 有几个概念 :
  • "leader" is the node responsible for all reads and writes for the given partition. Each node will be the leader for a randomly selected portion of the partitions.
  • "replicas" is the list of nodes that replicate the log for this partition regardless of whether they are the leader or even if they are currently alive.
  • "isr" is the set of "in-sync" replicas. This is the subset of the replicas list that is currently alive and caught-up to the leader.

    刚才那个topic

    bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test
    Topic:test PartitionCount:1 ReplicationFactor:1 Configs:
    Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0

发送 接收

  1. > bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
  2. ...
  3. my test message 1
  4. my test message 2
  5. ^C
  6. > bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
  7. ...
  8. my test message 1
  9. my test message 2
  10. ^C

试一下容错 fault-tolerance

  1. > ps aux | grep server-1.properties
  2. 7564 ttys002 0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.8/Home/bin/java...
  3. > kill -9 7564
  4. 看一下变化:Leader换了一个 因为1被干掉了
  5. > bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic my-replicated-topic
  6. Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
  7. Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 1,2,0 Isr: 2,0
  8. 还是收到了消息
  9. > bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
  10. ...
  11. my test message 1
  12. my test message 2
  13. ^C

七、使用kafka import/export data

刚才都是console 的数据,其他的sources other systems呢 用Kafka Connect

  1. 弄一个数据
  2. > echo -e "foo\nbar" > test.txt
  3. 启动 指定配置文件
  4. > bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties
  5. 验证一下
  6. > more test.sink.txt
  7. foo
  8. bar
  9. 消费者端
  10. > bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
  11. {"schema":{"type":"string","optional":false},"payload":"foo"}
  12. {"schema":{"type":"string","optional":false},"payload":"bar"}
  13. ...
  14. 可以继续写入
  15. > echo Another line>> test.txt

八、使用Kafka Streams

http://kafka.apache.org/22/documentation/streams/quickstart

WordCountDemo

https://github.com/apache/kafka/blob/2.2/streams/examples/src/main/java/org/apache/kafka/streams/examples/wordcount/WordCountDemo.java

代码片段

  1. // Serializers/deserializers (serde) for String and Long types
  2. final Serde<String> stringSerde = Serdes.String();
  3. final Serde<Long> longSerde = Serdes.Long();
  4. // Construct a `KStream` from the input topic "streams-plaintext-input", where message values
  5. // represent lines of text (for the sake of this example, we ignore whatever may be stored
  6. // in the message keys).
  7. KStream<String, String> textLines = builder.stream("streams-plaintext-input",
  8. Consumed.with(stringSerde, stringSerde);
  9. KTable<String, Long> wordCounts = textLines
  10. // Split each text line, by whitespace, into words.
  11. .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
  12. // Group the text words as message keys
  13. .groupBy((key, value) -> value)
  14. // Count the occurrences of each word (message key).
  15. .count()
  16. // Store the running counts as a changelog stream to the output topic.
  17. wordCounts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));

建一个 Kafka producer 指定input topic output topic

  1. > bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic streams-wordcount-output --config cleanup.policy=compact
  2. Created topic "streams-wordcount-output".

启动WordCount demo application

  1. bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo

启动一个生产者写数据

  1. > bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-plaintext-input
  2. all streams lead to kafka
  3. hello kafka streams

启动一个消费者接数据

  1. > bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic streams-wordcount-output --from-beginning --formatter kafka.tools.DefaultMessageFormatter --property print.key=true --property print.value=true --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer
  2. all 1
  3. streams 1
  4. lead 1
  5. to 1
  6. kafka 1
  7. hello 1
  8. kafka 2
  9. streams 2
  10. kafka 1

原文链接:http://www.cnblogs.com/tree1123/p/11150927.html

 友情链接:直通硅谷  点职佳  北美留学生论坛

本站QQ群:前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

W3xue 的所有内容仅供测试,对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接(目前9元/月)请联系QQ:27243702 沸活量
皖ICP备17017327号-2 皖公网安备34020702000426号