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<>();// 轉移阻塞隊列數據到普通的ListblockingQueue.drainTo(list);SQL = String.join("", list);System.out.println("SQL:"+SQL);endTime = System.currentTimeMillis();fifoCache.put("endTime",endTime);}}}

總結

以上是生活随笔為你收集整理的java消费kafka数据之后，进行堆积之后在插入数据库的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。