【项目实战】Kafka中Topic创建介绍

一、Kafka介绍

Kafka是一种分布式流处理平台，它可以处理实时数据流，支持高吞吐量、低延迟的数据处理。

它通过Topic和Partition机制将消息存储在集群中，并支持高吞吐量的消息发布和订阅。

二、Kafka中Topic介绍

2.1 可视为消息队列

Topic可以看作是一个消息队列

生产者将消息发送到Topic中，消费者从Topic中消费消息。

生产者将消息发布到Topic，而消费者从Topic订阅消息。

2.2 一种逻辑概念

在Kafka中，Topic是一种用于组织和存储消息的逻辑概念。

在Kafka中，Topic是一种逻辑概念，用于组织和管理消息。

2.3 与消息的关系

一个Topic可以被认为是一个特定的消息类别或者类型。

Kafka 中的 topic 就是消息流流动的载体，topic 是一个类似缓冲区的容器。

Kafka 会保存每个 topic 中的消息，发送到消费者的消息都可以分类放到相应的 topic 中。

每个消息都包含一个键和一个值，键用于标识消息，值是消息本身。

2.4 分区

Topic可以分为多个分区，每个分区可以在不同的机器上进行复制，以提高可靠性和容错性。

三、Kafka中Topic创建注意事项

创建一个Topic是在Kafka中使用的一项基本操作。

在使用Kafka时，需要先创建Topic，然后才能进行消息的发送和消费。

创建Topic时需要指定Topic的名称、分区数、副本数等参数。

• Topic在Kafka中扮演着非常重要的角色，影响了消息的处理和分发。
• 在Kafka中创建Topic时，名称应该是唯一的。

  为了保证高可用性和数据一致性，建议使用多个Partition和副本因子来创建Topic，并选择合适的消息分区策略和副本分配算法。同时，还应该合理设置Topic的配置参数，例如消息保留时间、压缩方式等，以满足不同的业务需求。此外，在生产环境中，建议使用Kafka的安全特性来保护Topic的访问。

  四、Kafka中创建Topic有两种方式

  在连接到Kafka集群后，

  可以使用Kafka提供的命令行工具或API来创建和管理Topic。

  4.1 命令行工具

  使用命令行工具kafka-topics.sh创建Topic。

  要使用此方法，您需要在安装Kafka时包含命令行工具。

  Kafka 用命令行界面创建 topic 时，需要指定参数包括：topic名称、partition 数、副本数。

  使用以下命令创建Topic：

  ./bin/kafka-create-topic.sh --zookeeper : --topic my-topic --partitions 3 --replication-factor 3
  ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic myTopic
  

  这将在本地Zookeeper实例上创建一个名为“myTopic”的Topic，使用单个副本和单个分区。

  您可以更改副本因子和分区数，具体取决于您的需求。

  4.2 API 编程方式- 使用 JavaAPI 创建Topic

  Kafka 提供了 java API 接口来进行topic的创建，通过KafkaAdminClient类的createTopics函数可以创建topic,提供需要创建的 Topic 名称以及该 topic 内的 partitions 个数、replicas 的个数等参数。

  代码示例：

  Properties props  = new Properties();
  props.put("bootstrap.servers", "localhost:9092");
  KafkaAdminClient adminClient = KafkaAdminClient.create(props);
  NewTopic newTopic = new NewTopic("my-topic", 3, (short)2);
  List list = new ArrayList<>();
  list.add(newTopic);
  adminClient.createTopics(list);
  adminClient.close();`
  

  使用Kafka的Java客户端API之一（如KafkaProducer或AdminClient）在代码中创建Topic。这种方法提供了更大的灵活性，因为您可以在代码中根据需求自定义Topic属性。

  以下是使用Kafka的Java客户端API创建Topic的示例代码：

  Properties properties = new Properties();
  properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  AdminClient adminClient = AdminClient.create(properties);
  NewTopic newTopic = new NewTopic("myTopic", 1, (short) 1);
  adminClient.createTopics(Collections.singleton(newTopic));
  

  五、实操：创建一个Topic的步骤

  5.1 连接Kafka集群

  首先，需要与Kafka集群建立连接。

  这可以通过Kafka提供的命令行工具（如kafka-console-producer.sh）或Kafka API完成。

  5.2 创建Topic

  在连接到Kafka集群后，可以使用Kafka提供的命令行工具或API创建Topic，需要指定Topic名称、Partition数量、副本因子等参数。

  创建Topic时需要指定名称、分区数量和复制因子等参数。

  分区数量指定Topic被分割为多少个分区，复制因子指定每个分区应该被复制到多少个Broker上。

  分区数量和复制因子的选择需要考虑可用的硬件资源和负载平衡等因素。

  5.3 发布和订阅消息

  创建Topic后，可以使用生产者将消息发布到Topic，消费者则可以从Topic订阅消息。

  • 发布消息：

    可以使用Producer API向指定的Topic发布消息，消息可以包含任意格式的数据，例如JSON、字符串或二进制数据等。

  • 订阅消息：

    可以使用Consumer API从指定的Topic订阅消息，Kafka支持两种消费方式：拉取（Pull）和推送（Push）。Pull方式需要手动调用API来获取消息，而Push方式则由Kafka自动推送消息给客户端。

  • 处理消息：

    可以使用Stream API来处理Kafka中的消息流，进行各种计算、转换、过滤等操作，并将结果发送到新的Topic中。

  • 监控和管理：

    Kafka提供了很多工具和API来监控和管理Topic，例如Kafka Manager、Kafka Monitor等，可以查看Topic的状态、偏移量、负载情况等信息。

    5.4 管理Topic

    Kafka提供了一些管理Topic的命令行工具，如kafka-topics.sh。

    这些工具可以用来列出所有的Topic，查看Topic的详细信息，增加或删除分区等操作。