概述
在使用kafka发送消息和消费消息之前,必须先要创建topic,在kafka中创建topic的方式有以下2种:
示例
举个demo:通过kafka-topics.sh脚本来创建一个名为topic-test1并且副本数为2、分区数为4的topic。(如无特殊说明,本文所述都是基于1.0.0版本。)
打开kafka-topics.sh脚本一探究竟,其内容只有一行,具体如下:
这个脚本的主要作用就是运行kafka.admin.TopicCommand。在main方法中判断参数列表中是否包含有"create“,如果有,那么就实施创建topic的任务。创建topic时除了需要zookeeper的地址参数外,还需要指定topic的名称、副本因子replication-factor以及分区个数partitions等必选参数 ,还可以包括disable-rack-aware、config、if-not-exists等可选参数。
createTopic详解
真正的创建过程是由createTopic这个方法中执行的,这个方法具体内容如下:
createTopic方法中首先获取topic的名称,config参数集以及判断是否有if-not-exists参数。config参数集可以用来设置topic级别的配置以覆盖默认配置。如果创建的topic再现有的集群中存在,那么会报出异常:TopicExistsException,如果创建的时候带了if-not-exists参数,那么发现topic冲突的时候可以不做任何处理;如果topic不冲突,那么和不带if-not-exists参数的行为一样正常topic,下面demo中接着上面的demo继续创建同一个topic,不带有if-not-exists参数和带有if-not-exists参数的效果如下:
接下去还会进一步检测topic名称中是否包含有“.”或者“_”字符的,这一个步骤在AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK()中调用validateCreateOrUpdateTopic()方法实现的。为什么要检测这两个字符呢?因为在Kafka的内部做埋点时会根据topic的名称来命名metrics的名称,并且会将句点号“.”改成下划线"_"。假设遇到一个topic的名称为“topic.1_2”,还有一个topic的名称为“topic_1.2”,那么最后的metrics的名称都为“topic_1_2”,所以就会发生名称冲突。举例如下,首先创建一个以"topic.1_2"名称的topic,提示WARNING警告,之后在创建一个“topic.1_2”时发生InvalidTopicException异常。
补充:topic的命名同样不推荐(虽然可以这样做)使用双下划线“__”开头,因为以双下划线开头的topic一般看作是kafka的内部topic,比如__consumer_offsets和__transaction_state。topic的名称必须由大小写字母、数字、“.”、“-”、“_”组成,不能为空、不能为“.”、不能为“..”,且长度不能超过249。
接下去createTopic方法的主体就分为两个部分了,如果检测出有replica-assignment参数,那么就是制定了副本的分配方式。这个在前面都没有提及,那么这个又指的是什么呢?如果包含了replica-assignment参数,那么就可以通过指定的分区副本分配方式创建topic,这个有点绕,不妨再来一个demo开拓下思路:
如果指定0:1, 0, 1:0这种(分区副本个数不同)就会报出AdminOperationException异常。详细demo如下:
当然,像0:1,,0:1,1:0这种企图跳过一个partition连续序号的行为也是不被允许的,详细demo如下:
验证之后在zookeeper中创建/brokers/topics/topic-test持久化节点,对应节点的数据就是以json格式呈现的分区分配的结果集,格式参考:{"version":1,"partitions":{"2":[0,1],"1":[1,0],"3":[1,0],"0":[0,1]}}。如果配置了config参数的话,同样先进行验证,如若无误就创建/config/topics/topic-test节点,并写入config对应的数据,格式参考:{"version":1,"config":{"max.message.bytes":"1000013"}}。详细demo如下:
更多内容
篇幅限制,更多内容将在下一篇中进行介绍,关注本微信公众号,了解更多细节。
领取专属 10元无门槛券
私享最新 技术干货