Kafka的单节点运行
启动服务
Kafka 使用 ZooKeeper 如果你还没有 ZooKeeper 服务器,你需要先启动一个 ZooKeeper 服务器。 您可以通过与 kafka 打包在一起的便捷脚本来快速简单地创建一个单节点 ZooKeeper 实例。如果你有使用docker 的经验,你可以使用 docker-compose 快速搭建一个 zk 集群。
bin/zookeeper-server-start.shconfig/zookeeper.properties
现在启动 Kafka 服务器:
> bin/kafka-server-start.sh config/server.properties
创建一个topic
创建一个名为“test”的 topic,它有一个分区和一个副本:
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
运行 list(列表)命令来查看这个 topic:
bin/kafka-topics.sh --list --zookeeper localhost:2181 test
除了手工创建 topic 外,你也可以配置你的 broker,当发布一个不存在的 topic 时自动创建 topic。
开启生产者 发送消息
Kafka 自带一个命令行客户端,它从文件或标准输入中获取输入,并将其作为 message(消息)发送到Kafka 集群。默认情况下,每行将作为单独的 message 发送。
运行 producer,然后在控制台输入一些消息以发送到服务器。
生产者不依赖zookeeper,都会发送给leader
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
启动消费者
Kafka 还有一个命令行使用者,它会将消息转储到标准输出。
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
如果在不同的终端中运行上述命令,能够在生产者终端中键入消息并看到它们出现在消费者终端中。
所有命令行工具都有选项; 运行不带参数的命令将显示使用信息
查看Topic信息
运行命令“describe topics” 查看集群中的 topic 信息
只有一个分区一个副本,Partition在broker0里面,后面的0都是borkerID
- “leader”是负责给定分区所有读写操作的节点。每个节点都是随机选择的部分分区的领导者。
- “replicas”是复制分区日志的节点列表,不管这些节点是 leader 还是仅仅活着。
- “isr”是一组“同步”replicas,是 replicas 列表的子集,它活着并被指到 leader。
Kafka Stream
Kafka Streams 是一个客户端库,用于构建任务关键型实时应用程序和微服务,其中输入和/或输出数据存储在 Kafka 集群中。Kafka Streams 结合了在客户端编写和部署标准 Java 和 Scala 应用程序的简单性以及 Kafka 服务器端集群技术的优势,使这些应用程序具有高度可扩展性,弹性,容错性,分布式等等。
以下是 WordCountDemo示例代码的要点(为了方便阅读,使用的是 java8 lambda 表达式)。
启动 zk 和 kafka
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties
准备输入主题并启动生产者
创建名为 streams-plaintext-input 的输入主题和名为 streams-wordcount-output 的输出主题:
> bin/kafka-topics.sh --create \
--zookeeper localhost:2181 \
--replication-factor 1 \
--partitions 1 \
--topic streams-plaintext-input
Created topic "streams-plaintext-input".
注意:我们创建输出主题并启用压缩,因为输
注意:我们创建输出主题并启用压缩,因为输出流是更改日志流
> bin/kafka-topics.sh --create \
--zookeeper localhost:2181 \
--replication-factor 1 \
--partitions 1 \
--topic streams-wordcount-output \
--config cleanup.policy=compact
Created topic "streams-wordcount-output".
使用相同的 kafka-topics 工具描述创建的主题:
bin/kafka-topics.sh --zookeeper localhost:2181 --describe
启动 Wordcount 应用程序
bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo
演示应用程序将从输入主题 stream-plaintext-input 读取,对每个读取消息执行 WordCount 算法的计算,并将其当前结果连续写入输出主题 streams-wordcount-output。
处理数据
开启一个生产者终端:
bin/kafka-console-producer.sh --broker-list localhost:9092 --topicstreams-plaintext-input
all streams lead to kafka
开启一个消费者终端:
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
这里,第一列是 java.lang.String 格式的 Kafka 消息键,表示正在计数的单词,第二列是 java.lang.Long
格式的消息值,表示单词的最新计数。
package com.study.kafka.stream.wordcount;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;
import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
/**
* Demonstrates, using the high-level KStream DSL, how to implement the WordCount program
* that computes a simple word occurrence histogram from an input text.
* <p>
* In this example, the input stream reads from a topic named "streams-plaintext-input", where the values of messages
* represent lines of text; and the histogram output is written to topic "streams-wordcount-output" where each record
* is an updated count of a single word.
* <p>
* Before running this example you must create the input topic and the output topic (e.g. via
* {@code bin/kafka-topics.sh --create ...}), and write some data to the input topic (e.g. via
* {@code bin/kafka-console-producer.sh}). Otherwise you won't see any data arriving in the output topic.
*/
public final class WordCountDemo {
public static void main(final String[] args) {
final Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.100.249:9092");
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
// setting offset reset to earliest so that we can re-run the demo code with the same pre-loaded data
// Note: To re-run the demo, you need to use the offset reset tool:
// https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> source = builder.stream("streams-plaintext-input");
final KTable<String, Long> counts = source
.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" ")))
.groupBy((key, value) -> value)
.count();
// need to override value serde to Long type
counts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));
final KafkaStreams streams = new KafkaStreams(builder.build(), props);
final CountDownLatch latch = new CountDownLatch(1);
// attach shutdown handler to catch control-c
Runtime.getRuntime().addShutdownHook(new Thread("streams-wordcount-shutdown-hook") {
@Override
public void run() {
streams.close();
latch.countDown();
}
});
try {
streams.start();
latch.await();
} catch (final Throwable e) {
System.exit(1);
}
System.exit(0);
}
}