提交最终代码

puzhibing

2026-08-21 3f6d5b788670170da195be8d8cee814069b3b224

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
package com.xiaopeng.automaticPilot.controller;
 
import com.alibaba.fastjson2.JSONObject;
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.RequestBody;
import org.springframework.web.bind.annotation.RestController;
 
@Slf4j
@RestController
public class KafkaTestController {
 
    @Autowired
    private KafkaConsumerService kafkaConsumerService;
 
    /**
     * Kafka消费 - XPU上报高优先级消息
     */
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-high",
            groupId = "group-robotaxi-xpu",
            containerFactory = "highPriorityFactory")
    public void consumeXpuHigh(String message) {
        log.info("收到XPU上报高优先级主题消息: {}", message);
        kafkaConsumerService.topicProdRobotaxiXpuHigh(message);
    }
 
    /**
     * Kafka消费 - XPU上报低优先级消息
     */
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-normal",
            groupId = "group-robotaxi-xpu",
            containerFactory = "normalPriorityFactory")
    public void consumeXpuNormal(String message) {
        log.info("收到XPU上报低优先级主题消息: {}", message);
        kafkaConsumerService.topicProdRobotaxiXpuNormal(message);
    }
 
    /**
     * HTTP测试接口 - 模拟高优先级消息
     */
    @PostMapping("/test/kafka/xpu-high")
    public void testXpuHigh(@RequestBody String message) {
        log.info("HTTP测试触发高优先级消息: {}", message);
        kafkaConsumerService.topicProdRobotaxiXpuHigh(message);
    }
 
    /**
     * HTTP测试接口 - 模拟低优先级消息
     */
    @PostMapping("/test/kafka/xpu-normal")
    public void testXpuNormal(@RequestBody String message) {
        log.info("HTTP测试触发低优先级消息: {}", message);
        kafkaConsumerService.topicProdRobotaxiXpuNormal(message);
    }
 
    /**
     * 模拟推送XPU消息
     */
    @PostMapping("/base/v1/pushXPUMessage")
    public R pushXPUMessage(@RequestBody String message) {
        log.info("收到XPU推送消息: {}", message);
        JSONObject json = new JSONObject();
        json.put("ts", System.currentTimeMillis());
        json.put("payload", message);
        kafkaConsumerService.topicProdRobotaxiXpuHigh(json.toJSONString());
        return R.ok();
    }
}