Spring-Kafka（三）—— 操作Topic以及Kafka

2018-09-07 本文已影响798人海苔胖胖

为什么要手动创建Topic

看过Spring-Kafka（二）这篇文章的三秒真男人会发现，我们从到到尾都没有创建过"topic.quick.demo"这个Topic，这是因为KafkaTemplate在发送的时候就已经帮我们完成了创建的操作，所以我们不需要主动创建"topic.quick.demo"这个Topic，而是交由KafkaTemplate去完成。但这样也出现了问题，这种情况创建出来的Topic的Partition(分区)数永远只有1个，也不会有副本（不知道的回炉重造，Kafka部署集群时使用的），这就导致了我们在后期不能顺利扩展。所以这种情况我们需要使用代码手动去创建Topic。

Kafka Tool 2

Kafka Tool 2是一款Kafka的可视化客户端工具，可以非常方便的查看Topic的队列信息以及消费者信息以及kafka节点信息。直接丢下载地址：http://www.kafkatool.com/download.html

打开之后我们看到的界面如下，非常简洁，虽然这个工具没有诸如KafkaOffsetMonitor这种监控工具的功能强大，但胜在操作方便，后期会补充一下监控工具的使用。

主界面
这个时候我们点击左上角的新建连接按钮去创建连接

创建连接
接下来会弹出一个配置连接的窗口，我们可以看到这个窗口左上角为Add Cluster(添加集群)，但没关系，对应单节点的Kafka实例来说也是可以的，因为这个软件监控的是Zookeeper而不是Kafka，Kafka的集群搭建也是依赖Zookeeper来实现的，所以默认情况下我们都是直接通过Zookeeper去完成大部分操作。

配置连接
点击Add后我们可以看到已经创建好的Topic。这个软件默认显示数据的类型为Byte，说白了我们是不能直接看到消息的明文数据的，没关系，我们可以在设置里面找到对应的修改选项，接下来就自己探索吧

界面

修改为明文显示

使用@Bean注解创建Topic

可以说使用SpringBoot创建Topic是一件非常简单的事情。
首先我们在config包下创建KafkaInitialConfiguration类，注册一个类型为NewTopic的Bean即可。

@Configuration
public class KafkaInitialConfiguration {

    //创建TopicName为topic.quick.initial的Topic并设置分区数为8以及副本数为1
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic("topic.quick.initial",8, (short) 1 );
    }
}

接下来启动一下SpringBoot项目，启动完成后打开Kafka Tool 2工具，查看一下刚才创建的队列是否存在

手动编码创建Topic

同样在KafkaInitialConfiguration类中编码，注册KafkaAdmin和AdminClient两个Bean

@Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> props = new HashMap<>();
        //配置Kafka实例的连接地址
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        KafkaAdmin admin = new KafkaAdmin(props);
        return admin;
    }

    @Bean
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin().getConfig());
    }

接下来在DemoTest测试类中编写测试方法，这里需要注意一点Topic的新增删除方法都是异步执行的，为了避免在创建过程中程序关闭导致创建失败，所以在代码最后加了一秒的休眠，执行测试方法我们打开Kafka Tool 2会发现多出了一个"topic.quick.initial2"的Topic

    @Autowired
    private AdminClient adminClient;

    @Test
    public void testCreateTopic() throws InterruptedException {
        NewTopic topic = new NewTopic("topic.quick.initial2", 1, (short) 1);
        adminClient.createTopics(Arrays.asList(topic));
        Thread.sleep(1000);
    }

创建结果

修改Topic分区数量

为什么要更新Topic呢，例如我们上一章创建的“topic.qucik.demo”只有一个分区，后期我们想增加分区数来提高系统吞吐量，这样我们就需要修改一下Topic的分区数了。实现也非常简单，只需要修改在我们刚才编写的KafkaInitialConfiguration类的initialTopic()方法，紧接着重启一下项目即可。修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小。

    @Bean
    public NewTopic initialTopic() {
        return new NewTopic("topic.quick.initial",8, (short) 1 );
    }
    //修改后|
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic("topic.quick.initial",11, (short) 1 );
    }

修改成功

查询Topic信息

测试类中新建一个testSelectTopicInfo方法，使用lambda表达式遍历输出

    @Test
    public void testSelectTopicInfo() throws ExecutionException, InterruptedException {
        DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList("topic.quick.initial"));
        result.all().get().forEach((k,v)->System.out.println("k: "+k+" ,v: "+v.toString()+"\n"));
    }

这个是输出结果，里面就包含了各个分区的信息等等

k: topic.quick.initial ,v: (name=topic.quick.initial,
    internal=false,
    partitions=(partition=0, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=1, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=2, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=3, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=4, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=5, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=6, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=7, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=8, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=9, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)),
            (partition=10, leader=admin-PC:9092 (id: 0 rack: null), replicas=admin-PC:9092 (id: 0 rack: null), isr=admin-PC:9092 (id: 0 rack: null)))