Apache Kafka 快速指南

ties)HelloMy first messageMy second message启动消费者以接收消息与生产者类似，在 config / consumer.proper-ties 文件中指定了缺省使用者属性。 打开一个新终端并键入以下消息消息语法。语法bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name --from-beginning示例bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-Kafka --from-beginning输出HelloMy first messageMy second message最后，您可以从制作商的终端输入消息，并看到他们出现在消费者的终端。 到目前为止，您对具有单个代理的单节点群集有非常好的了解。 现在让我们继续讨论多个代理配置。单节点多代理配置在进入多个代理集群设置之前，首先启动ZooKeeper服务器。创建多个Kafka Brokers - 我们在配置/ server.properties中已有一个Kafka代理实例。 现在我们需要多个代理实例，因此将现有的server.prop-erties文件复制到两个新的配置文件中，并将其重命名为server-one.properties和server-two.properties。 然后编辑这两个新文件并分配以下更改 - config / server-one.properties# The id of the broker. This must be set to a unique integer for each broker.broker.id=1# The port the socket server listens onport=9093# A comma seperated list of directories under which to store log fileslog.dirs=/tmp/kafka-logs-1config / server-two.properties# The id of the broker. This must be set to a unique integer for each broker.broker.id=2# The port the socket server listens onport=9094# A comma seperated list of directories under which to store log fileslog.dirs=/tmp/kafka-logs-2启动多个代理 - 在三台服务器上进行所有更改后，打开三个新终端，逐个启动每个代理。Broker1bin/kafka-server-start.sh config/server.propertiesBroker2bin/kafka-server-start.sh config/server-one.propertiesBroker3bin/kafka-server-start.sh config/server-two.properties现在我们有三个不同的经纪人在机器上运行。 自己尝试，通过在ZooKeeper终端上键入 jps 检查所有守护程序，然后您将看到响应。创建主题让我们为此主题将复制因子值指定为三个，因为我们有三个不同的代理运行。 如果您有两个代理，那么分配的副本值将是两个。语法bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 -partitions 1 --topic topic-name示例bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 -partitions 1 --topic Multibrokerapplication输出created topic “Multibrokerapplication" Describe 命令用于检查哪个代理正在侦听当前创建的主题，如下所示 - bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic Multibrokerappli-cation输出bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic Multibrokerappli-cationTopic:Multibrokerapplication PartitionCount:1 ReplicationFactor:3 Configs: Topic:Multibrokerapplication Partition:0 Leader:0 Replicas:0,2,1 Isr:0,2,1从上面的输出，我们可以得出结论，第一行给出所有分区的摘要，显示主题名称，分区数量和我们已经选择的复制因子。 在第二行中，每个节点将是分区的随机选择部分的领导者。在我们的例子中，我们看到我们的第一个broker(with broker.id 0)是领导者。 然后Replicas:0,2,1意味着所有代理复制主题最后 Isr 是 in-sync 副本的集合。 那么，这是副本的子集，当前活着并被领导者赶上。启动生产者以发送消息此过程保持与单代理设置中相同。示例bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Multibrokerapplication输出bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Multibrokerapplication[2016-01-20 19:27:21,045] WARN Property topic is not valid (kafka.utils.Verifia-bleProperties)This is single node-multi broker demoThis is the second message启动消费者以接收消息此过程保持与单代理设置中所示的相同。示例bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Multibrokerapplica-tion --from-beginning输出bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Multibrokerapplica-tion —from-beginningThis is single node-multi broker demoThis is the second message基本主题操作在本章中，我们将讨论各种基本主题操作。修改主题您已经了解如何在Kafka Cluster中创建主题。 现在让我们使用以下命令修改已创建的主题语法bin/kafka-topics.sh —zookeeper localhost:2181 --alter --topic topic_name --parti-tions count示例We have already created a topic “Hello-Kafka" with single partition count and one replica factor. Now using “alter" command we have changed the partition count.bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic Hello-kafka --parti-tions 2输出WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affectedAdding partitions succeeded!删除主题要删除主题，可以使用以下语法。语法bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic topic_name示例bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic Hello-kafka输出> Topic Hello-kafka marked for deletion注意 - 如果 delete.topic.enable 未设置为true，则此操作不会产生任何影响Apache Kafka - 简单生产者示例让我们使用Java客户端创建一个用于发布和使用消息的应用程序。 Kafka生产者客户端包括以下API。KafkaProducer API让我们了解本节中最重要的一组Kafka生产者API。 KafkaProducer API的中心部分是 KafkaProducer 类。 KafkaProducer类提供了一个选项，用于将其构造函数中的Kafka代理连接到以下方法。KafkaProducer类提供send方法以异步方式将消息发送