[大数据] java消费kafka数据之后，进行堆积之后在插入数据库

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> java消费kafka数据之后，进行堆积之后在插入数据库 -> 正文阅读

[大数据]java消费kafka数据之后，进行堆积之后在插入数据库

java高频的获取kafka数据，导致数据库数据一致在高频读写，为了降低数据库的高频连接搞高频读写，可以将数据堆积一段时间之后，进行插入数据库操作。

主要采用了队列和缓存，将获取到的数据放入java队列中，利用缓存进行延时判断。

 <dependency>
      <groupId>cn.hutool</groupId>
      <artifactId>hutool-all</artifactId>
      <version>5.8.0.M2</version>
    </dependency>


    <dependency>
      <groupId>com.google.guava</groupId>
      <artifactId>guava</artifactId>
      <version>26.0-jre</version>
    </dependency>

    @KafkaListener(topics = {"jyz_xxxxxxx"})
    public void jyz_xxxxxxx(ConsumerRecord<?, ?> record) throws InterruptedException {

        Optional<?> kafkaMessage = Optional.ofNullable(record.value());

        String  SQL=null;

        if (kafkaMessage.isPresent()) {

            Object message = kafkaMessage.get();

            blockingQueue.offer(message.toString());

            long startTime= System.currentTimeMillis();

            long endTime = 0;

            try {
                endTime = fifoCache.get("endTime");
            } catch (Exception e) {
                endTime = System.currentTimeMillis();

                fifoCache.put("endTime",endTime);
            }

            if(startTime-endTime>=2000){

                List<String> list = new ArrayList<>();
                // 转移阻塞队列数据到普通的List
                blockingQueue.drainTo(list);

                 SQL = String.join("", list);

                System.out.println("SQL:"+SQL);

                endTime = System.currentTimeMillis();

                fifoCache.put("endTime",endTime);


            }



        }
    }