通过前面两篇，了解了Kafka的基本概念的安装配置，现在在Java中调用它的API来初步实现一些功能。

        先备条件：

        - ZooKeeper已启动

        - Kafka实例代理已启动，为了方便查看，也启动Kafka Manager网站，如下，有一个名为jim_test，id为0的broker，里面有一个默认的主题:

        开始……

        1. IDEA新建一个用maven管理的Java项目

        2. pom.xml引入kafka的依赖包

        3. 主题基本操作

            3.1 查看主题

package cn.linjk.kafka;

import kafka.admin.AdminUtils;
import kafka.utils.ZkUtils;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.security.JaasUtils;

public class Main {
    private static final String ZK_HOST = "localhost:2181"; // 集群时: "localhost-1:2181,localhsot-2:2181"
    private static final int TIMEOUT_SESSION = 3000;
    private static final int TIMEOUT_CONNECT = 3000;

    public static void main(String[] args) {
        topic_info("__consumer_offsets");
    }

    private static void topic_info(String topic_name) {
        ZkUtils zkUtils = null;
        try {
            zkUtils = ZkUtils.apply(ZK_HOST, TIMEOUT_SESSION, TIMEOUT_CONNECT,
                    JaasUtils.isZkSecurityEnabled());
            MetadataResponse.TopicMetadata topicMetadata =  AdminUtils.fetchTopicMetadataFromZk(
                    topic_name, zkUtils);
            System.out.println(topicMetadata.partitionMetadata().size());
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        finally {
            zkUtils.close();
        }
    }
}

            3.2 创建主题

                  创建主题时，需要指定主题的分区数和副本数，同时可以设置主题的级别，还可以执行机架感知类型(主要针对自动分配副本策略时指定代理机架信息)。

                  可以看到，创建主题是没返回的，只要没异常，主题都是会创建成果的。

            3.3 修改主题配置

                  配置的这次修改会覆盖上一次修改，如果本次修改没有传上次配置的参数，上次配置的参数会恢复到默认值，因此建议修改前先获取上次配置，然后在上次配置的基础上进行修改。

            3.4 删除主题

                   可以看到，并未真正删除主题，只是标记其状态为删除，如果需要彻底删除，需要手动删除相应文件及节点，如果想直接可以删除，需要修改Kafka的server.properties配置文件以下属性为true:

        4. 小结

            在上面的查看、新增、修改和删除中，可以看到两个重要的类: ZkUtils和AdminUtils，可以对应源码看其作用和进行实际测试即可。