|
@@ -0,0 +1,62 @@
|
|
|
+package com.css.simulation.resource.scheduler.controller;
|
|
|
+
|
|
|
+import api.common.pojo.dto.ProjectMessageDTO;
|
|
|
+import api.common.util.JsonUtil;
|
|
|
+import com.css.simulation.resource.scheduler.consumer.ManualProjectConsumer;
|
|
|
+import lombok.SneakyThrows;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.kafka.core.KafkaTemplate;
|
|
|
+import org.springframework.web.bind.annotation.PostMapping;
|
|
|
+import org.springframework.web.bind.annotation.RequestBody;
|
|
|
+import org.springframework.web.bind.annotation.RequestMapping;
|
|
|
+import org.springframework.web.bind.annotation.RestController;
|
|
|
+
|
|
|
+@RequestMapping("/kafka")
|
|
|
+@RestController
|
|
|
+@Slf4j
|
|
|
+public class KafkaController {
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ KafkaTemplate<String, String> kafkaTemplate;
|
|
|
+ @Autowired
|
|
|
+ ManualProjectConsumer manualProjectConsumer;
|
|
|
+
|
|
|
+ @PostMapping("/hello")
|
|
|
+ public void hello() {
|
|
|
+ kafkaTemplate.send("hello", "hello world!").addCallback(success -> {
|
|
|
+
|
|
|
+ String topic = success.getRecordMetadata().topic();
|
|
|
+
|
|
|
+ int partition = success.getRecordMetadata().partition();
|
|
|
+
|
|
|
+ long offset = success.getRecordMetadata().offset();
|
|
|
+ log.info("------- 发送消息成功:\n"
|
|
|
+ + "主题 topic 为:" + topic + "\n"
|
|
|
+ + "分区 partition 为:" + partition + "\n"
|
|
|
+ + "偏移量为:" + offset);
|
|
|
+ }, failure -> {
|
|
|
+ log.error("------- 发送消息失败:" + failure.getMessage());
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ @PostMapping("/send")
|
|
|
+ @SneakyThrows
|
|
|
+ public void send(@RequestBody ProjectMessageDTO projectMessageDTO) {
|
|
|
+ kafkaTemplate.send(projectMessageDTO.getProjectId(), JsonUtil.beanToJson(projectMessageDTO)).addCallback(success -> {
|
|
|
+
|
|
|
+ String topic = success.getRecordMetadata().topic();
|
|
|
+
|
|
|
+ int partition = success.getRecordMetadata().partition();
|
|
|
+
|
|
|
+ long offset = success.getRecordMetadata().offset();
|
|
|
+ log.info("------- 发送消息成功:\n"
|
|
|
+ + "主题 topic 为:" + topic + "\n"
|
|
|
+ + "分区 partition 为:" + partition + "\n"
|
|
|
+ + "偏移量为:" + offset);
|
|
|
+ }, failure -> {
|
|
|
+ log.error("------- 发送消息失败:" + failure.getMessage());
|
|
|
+ });
|
|
|
+ }
|
|
|
+}
|