Java程序创建Kafka Topic，以及数据生产消费，常用的命令

Posted 2021-12-09 PacosonSWJTU

转自： Java程序创建Kafka Topic，以及数据生产消费，常用的命令_Zyy_z_的博客-CSDN博客_java kafka创建topicKafka简介： Kafka是一个分布式发布——订阅消息传递系统。Kafka快速、可扩展且耐用。它保留主题中的消息源。生产者将数据写入主题，消费者从主题中读取数据。Kafka的特点： 1. 同时为分布和订阅提供高吞吐量。据了解，Kafka每秒可以生产约25万条消息（50MB），每秒处理55万条消息...https://blog.csdn.net/Zyy_z_/article/details/101680138

---

【1】Kafka简介

Kafka是一个分布式发布——订阅消息传递系统。Kafka快速、可扩展且耐用。它保留主题中的消息源。生产者将数据写入主题，消费者从主题中读取数据。

1）Kafka的特点：

• 1. 同时为分布和订阅提供高吞吐量。 据了解，Kafka每秒可以生产约25万条消息（50MB），每秒处理55万条消息（110MB）这里说条数，可能不上特别准确，因为消息的大小可能不一致；
• 2. 可进行持久化操作，将消息持久化到到磁盘，以日志的形式存储，因此可用于批量消费，例如ETL，以及实时应用程序。 通过将数据持久化到硬盘以及replication防止数据丢失。
• 3. 分布式系统，易于向外拓展。所有的Producer、broker和consumer都会有多个，均为分布式。无需停机即可拓展 机器。
• 4. 消息被处理的状态是在consumer端维护，而不是由server端维护，当失败时能自动平衡。

2）Kafka名词解释：

1. producer：消息的生成者
2. consumer：消息的消费者
3. topic：你把它理解为标签
4. broker：Kafka处理资源的消息源(feeds of messages)的不同分类

3）Kafka常用命令:

1. 创建主题（4个分区，2个副本):  kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 4 --topic test
2. 查询所有Topic：kafka-topics.sh --zookeeper localhost:2181 --list
3. 查看指定得Topic：kafka-topics.sh --zookeeper localhost:2181 --describe --topic t_cdr
4. 删除Topic：kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper localhost:2181 --topic t_cdr
5. 生产者 ：kafka-console-producer.sh --broker-list localhost:9092 --topic test
6. 消费者 : kafka-console-consumer.sh --zookeeper localhost:2181 --topic test
7. 新生产者（支持0.9版本+）:  kafka-console-producer.sh --broker-list localhost:9092 --topic test --producer.config config/producer.properties
8. 新消费者（支持0.9版本+）:  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --new-consumer --from-beginning --consumer.config config/consumer.properties

---

【2】kafka java api

【2.1】Java程序操作创建Topic：   

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all"); props.put("retries", 0);
props.put("batch.size", 16384); props.put("linger.ms", 1);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
AdminClient  create = KafkaAdminClient.create(props);//创建Topic
create.createTopics(Lists.newArrayList(new NewTopic("Topic名称")，1，（short）1));//一个分区
create.close();//关闭

其他创建Topic得方式Java API：

https://blog.csdn.net/meng984611383/article/details/80500761https://blog.csdn.net/meng984611383/article/details/80500761

---

【2.2】Kafka生产数据:

Producer<String, String> producer = new KafkaProducer<>(props);
for(int i = 0; i < 100; i++) //生产数据
    producer.send(new ProducerRecord<String, String>("Topic名称", Integer.toString(i), Integer.toString(i)));
producer.close(); //关闭

---

【2.3】消费数据

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo", "bar"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

生产者的缓冲空间池保留尚未发送到服务器的消息，后台I/O线程负责将这些消息转换成请求发送到集群。如果使用后不关闭生产者，则会泄露这些资源。

send()方法是异步的，添加消息到缓冲区等待发送，并立即返回。生产者将单个的消息批量在一起发送来提高效率。

1. ack是判别请求是否为完整的条件（就是是判断是不是成功发送了）。我们指定了“all”将会阻塞消息，这种设置性能最低，但是是最可靠的。
2. retries，如果请求失败，生产者会自动重试，我们指定是0次，如果启用重试，则会有重复消息的可能性。
3. producer(生产者)缓存每个分区未发送的消息。缓存的大小是通过 batch.size 配置指定的。值较大的话将会产生更大的批。并需要更多的内存（因为每个“活跃”的分区都有1个缓冲区）。
4. 默认缓冲可立即发送，即便缓冲空间还没有满，但是，如果你想减少请求的数量，可以设置linger.ms大于0。这将指示生产者发送请求之前等待一段时间，希望更多的消息填补到未满的批中。这类似于TCP的算法，例如上面的代码段，可能100条消息在一个请求发送，因为我们设置了linger(逗留)时间为1毫秒，然后，如果我们没有填满缓冲区，这个设置将增加1毫秒的延迟请求以等待更多的消息。需要注意的是，在高负载下，相近的时间一般也会组成批，即使是 linger.ms=0。在不处于高负载的情况下，如果设置比0大，以少量的延迟代价换取更少的，更有效的请求。
5. buffer.memory 控制生产者可用的缓存总量，如果消息发送速度比其传输到服务器的快，将会耗尽这个缓存空间。当缓存空间耗尽，其他发送调用将被阻塞，阻塞时间的阈值通过max.block.ms设定，之后它将抛出一个TimeoutException。
6. key.serializer和value.serializer示例，将用户提供的key和value对象ProducerRecord转换成字节，你可以使用附带的ByteArraySerializaer或StringSerializer处理简单的string或byte类型。