zoukankan      html  css  js  c++  java
  • Spring Kafka(二)操作Topic以及Kafka Tool 2的使用

    1. 为什么要手动创建Topic

    我们从到到尾都没有创建过"topic.quick.demo"这个Topic,这是因为KafkaTemplate在发送的时候就已经帮我们完成了创建的操作

    ,所以我们不需要主动创建"topic.quick.demo"这个Topic,而是交由KafkaTemplate去完成。

    但这样也出现了问题,这种情况创建出来的Topic的Partition(分区)数永远只有1个,也不会有副本(不知道的回炉重造,Kafka部署集群时使用的),

    这就导致了我们在后期不能顺利扩展。所以这种情况我们需要使用代码手动去创建Topic。

    2. Kafka Tool 2

    Kafka Tool 2是一款Kafka的可视化客户端工具,可以非常方便的查看Topic的队列信息以及消费者信息以及kafka节点信息。

    直接丢下载地址:http://www.kafkatool.com/download.html

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

    3. 使用@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 );
        }
    }
    View Code

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

    4. 手动编码创建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());
        }
    View Code

    接下来在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);
        }
    View Code

    5. 修改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 );
        }
    View Code

    6. 查询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"));
        }
    View Code

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

    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)))
    View Code

     

  • 相关阅读:
    Linux 分区与挂载
    Linux Shell 编程总结
    Java AtomicIntegerFieldUpdater 抽象类
    Java 内部类加载顺序
    Java AtomicIntegerArray 类
    Java AtomicInteger 类
    Java ReentrantLock 类
    Java Runnable 接口
    Java Supplier 接口
    Java Consumer 接口
  • 原文地址:https://www.cnblogs.com/yanliang12138/p/12492498.html
Copyright © 2011-2022 走看看