diff --git a/src/main/java/com/example/zxweb/controller/KafkaController.java b/src/main/java/com/example/zxweb/controller/KafkaController.java index 9ede275..ff5de9d 100644 --- a/src/main/java/com/example/zxweb/controller/KafkaController.java +++ b/src/main/java/com/example/zxweb/controller/KafkaController.java @@ -51,6 +51,27 @@ public class KafkaController { return Result.OK(test); } + @GetMapping("/consumer2/{topic}") + public Result consumer2(@PathVariable String topic){ + if (StringUtils.isEmpty(topic)) { + return Result.error("请确认topic是否正确"); + } + List resultList = new ArrayList<>(); + Consumer consumer = new KafkaConsumer<>(kafkaUtil.getSafeConsumerPro()); + consumer.subscribe(Collections.singletonList(topic)); + ConsumerRecords records = consumer.poll(5000); + JSONArray jsonArray=new JSONArray(); + for (ConsumerRecord consumerRecord : records) { + String kfkContent = consumerRecord.value(); + resultList.add(kfkContent); + if (!StringUtils.isEmpty(kfkContent)) { + jsonArray.add(JSONObject.parseObject(kfkContent)); + } + } + consumer.close(); + return Result.OK(jsonArray); + } + @GetMapping("/consumer/{topic}") @ApiOperation(value = "consumer") public Result getConsumerData(@PathVariable String topic) {