package com.xiaopeng.automaticPilot.controller; import com.xiaopeng.automaticPilot.service.KafkaConsumerService; import com.xiaopeng.common.core.domain.R; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RestController; @Slf4j @RestController public class KafkaTestController { @Autowired private KafkaConsumerService kafkaConsumerService; /** * 消费消息 - XPU上报Robotaxi云端的消息topic(高优先级) * * @param message 消息内容 */ @PostMapping("/test/kafka/xpu-high") @KafkaListener(topics = "topic-prod-robotaxi-xpu-high", groupId = "group-robotaxi-xpu") public void topicProdRobotaxiXpuHigh(String message) { log.info("收到XPU上报高优先级主题消息: {}", message); kafkaConsumerService.topicProdRobotaxiXpuHigh(message); } /** * 消费消息 - XPU上报Robotaxi云端的消息topic(低优先级) * * @param message 消息内容 */ @PostMapping("/test/kafka/xpu-normal") @KafkaListener(topics = "topic-prod-robotaxi-xpu-normal", groupId = "group-robotaxi-xpu") public void topicProdRobotaxiXpuNormal(String message) { log.info("收到XPU上报低优先级主题消息: {}", message); kafkaConsumerService.topicProdRobotaxiXpuNormal(message); } /** * 模拟推送XPU消息 * @param message * @return */ @PostMapping("/base/v1/pushXPUMessage") public R pushXPUMessage(String message) { log.info("收到XPU推送消息: {}", message); kafkaConsumerService.topicProdRobotaxiXpuHigh(message); return R.ok(); } }