xp-robotaxi-api/src/main/java/com/xiaopeng/api/dto/automaticPilot/VehicleEventDTO.java
New file @@ -0,0 +1,22 @@ package com.xiaopeng.api.dto.automaticPilot; import io.swagger.annotations.ApiModelProperty; import lombok.Data; import java.util.Map; @Data public class VehicleEventDTO { @ApiModelProperty(value = "车辆VIN") private String vin; @ApiModelProperty(value = "XPU ID") private String xpuId; @ApiModelProperty(value = "订单ID") private String orderId; @ApiModelProperty(value = "事件标识") private String eventKey; @ApiModelProperty(value = "事件时间") private Long eventTime; @ApiModelProperty(value = "事件属性") private Map<String,Object> attributes; } xp-robotaxi-api/src/main/java/com/xiaopeng/api/factory/automaticPilot/VehicleBuriedPointClientFallbackFactory.java
New file @@ -0,0 +1,23 @@ package com.xiaopeng.api.factory.automaticPilot; import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO; import com.xiaopeng.api.feignClient.automaticPilot.VehicleBuriedPointClient; import com.xiaopeng.api.feignClient.automaticPilot.XpuVinClient; import com.xiaopeng.common.core.domain.R; import lombok.extern.slf4j.Slf4j; import org.springframework.cloud.openfeign.FallbackFactory; @Slf4j public class VehicleBuriedPointClientFallbackFactory implements FallbackFactory<VehicleBuriedPointClient> { @Override public VehicleBuriedPointClient create(Throwable cause) { return new VehicleBuriedPointClient() { @Override public R<String> sendVehicleEvent(VehicleEventDTO event) { log.error("推送埋点数据失败: {}", cause.getMessage()); return R.fail("推送埋点数据失败"); } }; } } xp-robotaxi-api/src/main/java/com/xiaopeng/api/feignClient/automaticPilot/VehicleBuriedPointClient.java
New file @@ -0,0 +1,27 @@ package com.xiaopeng.api.feignClient.automaticPilot; import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO; import com.xiaopeng.api.factory.automaticPilot.VehicleBuriedPointClientFallbackFactory; import com.xiaopeng.api.factory.automaticPilot.XpuVimClientFallbackFactory; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.dto.ServiceNameConstantsEnum; import org.springframework.cloud.openfeign.FeignClient; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestParam; @FeignClient(contextId = "VehicleBuriedPointClient", value = ServiceNameConstantsEnum.XP_AUTOMATIC_PILOT_SERVICE, fallbackFactory = VehicleBuriedPointClientFallbackFactory.class) public interface VehicleBuriedPointClient { /** * 推送埋点信息 * @param event * @return */ @PostMapping("/inner/vehicleBuriedPoint/sendVehicleEvent") R<String> sendVehicleEvent(@RequestBody VehicleEventDTO event); } xp-robotaxi-api/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
@@ -32,8 +32,8 @@ com.xiaopeng.api.factory.automaticPilot.XPUClientFallbackFactory com.xiaopeng.api.factory.automaticPilot.XpuVimClientFallbackFactory com.xiaopeng.api.factory.automaticPilot.OrderMissionClientFallbackFactory com.xiaopeng.api.factory.automaticPilot.VehicleBuriedPointClientFallbackFactory com.xiaopeng.api.factory.maintenance.XpCheckLocationClientFallbackFactory com.xiaopeng.api.factory.automaticPilot.OrderMissionClientFallbackFactory com.xiaopeng.api.factory.maintenance.XpClockInClientFallbackFactory com.xiaopeng.api.factory.account.SysGroupMenuClientFallbackFactory com.xiaopeng.api.factory.order.XpCruiseOrderClientFallbackFactory xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/VehicleBuriedPointController.java
New file @@ -0,0 +1,31 @@ package com.xiaopeng.automaticPilot.controller; import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO; import com.xiaopeng.automaticPilot.util.VehicleBuriedPointProducer; import com.xiaopeng.common.core.domain.R; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.*; @Slf4j @RestController @RequestMapping("") public class VehicleBuriedPointController { @Autowired private VehicleBuriedPointProducer vehicleBuriedPointProducer; /** * 推送埋点信息 * @param event * @return */ @PostMapping("/inner/vehicleBuriedPoint/sendVehicleEvent") public R<String> sendVehicleEvent(@RequestBody VehicleEventDTO event){ vehicleBuriedPointProducer.sendVehicleEvent(event); return R.ok(); } } xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/KafkaSendUtils.java
New file @@ -0,0 +1,84 @@ package com.xiaopeng.automaticPilot.util; import io.growing.sdk.java.GrowingAPI; import io.growing.sdk.java.dto.GIOMessage; import io.growing.sdk.java.dto.GioCdpEventMessage; import java.util.Arrays; import java.util.HashMap; import java.util.Map; public class KafkaSendUtils { private static final GrowingAPI project = new GrowingAPI.Builder().setProjectKey("89a922449121f7c2").setDataSourceId("9365c3d2f397f172").build(); /** * * 车辆状态变更 触发时机：车辆主状态或子状态发生变化的瞬间。 * 发送数据到GrowingIO * * @param map */ public static void sendVehicleStatusChanged(Map<String, Object> map) { // Config GrowingIO // 参数需要从CDP增长平台上，创建新应用，或从已创建的数据源中获取, 如不清楚请联系您的专属项目经理 // YourAccountId eg: 89a922449121f7c2 // YourDatasourceId eg: 9365c3d2f397f172 //事件行为消息体，anonymousId 和 loginUserId 参数，不能同时为空 GioCdpEventMessage eventMessage = new GioCdpEventMessage.Builder() .eventTime(System.currentTimeMillis()) // 默认为系统当前时间 (选填) .eventKey("RT_web_vehicle_status_changed") // 事件标识 (必填) // .domain("com.growingio.app") // App包名或H5域名（选填） // .urlScheme("growing.123c12fb12f123cc") // 链接协议（选填） // .deviceBrand("google") // 设备品牌（选填） // .deviceModel("Nexus 5") // 设备型号（选填） // .deviceType("PHONE") // 设备类型（选填） // .appVersion("1.2.4") // App版本（选填） // .appName("看数助手") // App名称（选填） .language("zh_CN") // 语言（选填） .anonymousId("device_id") // 访问用户ID (选填) // .loginUserKey("account") // 登录用户KEY (选填，需有规划并在平台配置后再上报) .loginUserId("417abcabcabcbac") // 登录用户ID (选填) // .addEventVariable("product_name", "cdp苹果") // 事件属性 (选填) // .addEventVariable("product_classify", Arrays.asList("苹果", "香蕉")) // 事件属性 (选填) .addEventVariables(map) // 事件属性集合 (选填) // .addItem("itemId", "itemKey") // 物品模型ID, KEY (选填) .build(); //上传事件行为消息到服务器 project.send(eventMessage); } /** * * 订单状态变更 * 开始接单，上报一次 * 订单结束，上报一次 * 发送数据到GrowingIO * * @param map */ public static void sendVehicleTraceUpload(Map<String, Object> map) { //事件行为消息体，anonymousId 和 loginUserId 参数，不能同时为空 GioCdpEventMessage eventMessage = new GioCdpEventMessage.Builder() .eventTime(System.currentTimeMillis()) // 默认为系统当前时间 (选填) .eventKey("RT_xpu_vehicle_trace_upload") // 事件标识 (必填) .language("zh_CN") // 语言（选填） .addEventVariables(map) // 事件属性集合 (选填) .build(); //上传事件行为消息到服务器 project.send(eventMessage); } public static void main(String[] args) { Map<String, Object> map = new HashMap<>(); map.put("vin", "L1NSPGHB7SB018421"); map.put("order_id", ""); map.put("timestamp", System.currentTimeMillis()); map.put("sub_status", ""); map.put("prev_sub_status", ""); sendVehicleStatusChanged(map); } } xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/VehicleBuriedPointProducer.java
New file @@ -0,0 +1,161 @@ package com.xiaopeng.automaticPilot.util; import com.fasterxml.jackson.databind.ObjectMapper; import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.producer.*; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import java.util.HashMap; import java.util.Map; import java.util.Properties; import java.util.concurrent.Future; @Slf4j @Component public class VehicleBuriedPointProducer { // 预发环境 Kafka 集群 @Value("${spring.kafka.bootstrap-servers}") private String bootstrapServers; // 目标 Topic private static final String TOPIC = "topic-robotaxi-vehicle-buriedPoint"; private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); private Properties kafkaProperties; @PostConstruct public void init() { kafkaProperties = buildProperties(); } /** * 发送埋点消息 * * @param event * @throws Exception */ public void sendVehicleEvent(VehicleEventDTO event){ if (kafkaProperties == null) { kafkaProperties = buildProperties(); } KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProperties); try { // 转 JSON String messageBody = OBJECT_MAPPER.writeValueAsString(event); // 建议用 vin 作为 key（保证同车有序） ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, event.getVin(), messageBody); Future<RecordMetadata> future = producer.send(record); RecordMetadata metadata = future.get(); log.info("发送成功："); log.info("Topic: " + metadata.topic()); log.info("Partition: " + metadata.partition()); log.info("Offset: " + metadata.offset()); } catch (Exception e) { log.error("发送失败：{}", e.getMessage()); e.printStackTrace(); } finally { producer.close(); } } /** * 构造埋点消息体 * * @param vin * @param orderId * @param xpuId * @param eventKey * @param eventTime * @param extraAttributes * @return */ public VehicleEventDTO buildVehicleEvent(String vin, String orderId, String xpuId, String eventKey, Long eventTime, Map<String, Object> extraAttributes) { VehicleEventDTO event = new VehicleEventDTO(); event.setVin(vin); event.setXpuId(xpuId != null ? xpuId : ""); event.setOrderId(orderId != null ? orderId : ""); event.setEventKey(eventKey); event.setEventTime(eventTime != null ? eventTime : System.currentTimeMillis()); Map<String, Object> attributes = new HashMap<>(); attributes.put("vin", vin); attributes.put("order_id", orderId != null ? orderId : ""); attributes.put("timestamp", event.getEventTime()); if (extraAttributes != null) { attributes.putAll(extraAttributes); } event.setAttributes(attributes); return event; } private Properties buildProperties() { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // 生产建议配置 props.put(ProducerConfig.ACKS_CONFIG, "all"); props.put(ProducerConfig.RETRIES_CONFIG, 3); props.put(ProducerConfig.LINGER_MS_CONFIG, 5); props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); return props; } // public static void main(String[] args) throws Exception { // // long timeMillis = System.currentTimeMillis(); // // // 构造埋点消息体（示例） // VehicleEvent event = new VehicleEvent(); // event.setVin("L1NSPGHB7SB018421"); // event.setXpuId(""); // event.setOrderId(""); // event.setEventKey("RT_web_vehicle_status_changed"); // event.setEventTime(timeMillis); // // Map<String, Object> attributes = new HashMap<>(); // attributes.put("vin", "L1NSPGHB7SB018421"); // attributes.put("order_id", ""); // attributes.put("timestamp", timeMillis); // attributes.put("sub_status", ""); // attributes.put("prev_sub_status", ""); // event.setAttributes(attributes); // // // 转 JSON // String messageBody = OBJECT_MAPPER.writeValueAsString(event); // // // 建议用 vin 作为 key（保证同车有序） // ProducerRecord<String, String> record = // new ProducerRecord<>(TOPIC, event.getVin(), messageBody); // // try { // Future<RecordMetadata> future = producer.send(record); // RecordMetadata metadata = future.get(); // // System.out.println("发送成功："); // System.out.println("Topic: " + metadata.topic()); // System.out.println("Partition: " + metadata.partition()); // System.out.println("Offset: " + metadata.offset()); // // } catch (Exception e) { // e.printStackTrace(); // } finally { // producer.close(); // } // } } xp-robotaxi-automatic-pilot-boot/src/main/resources/gio.properties
New file @@ -0,0 +1,26 @@ #项目采集端地址 api.host=https://collect.test.xiaopeng.com #消息发送间隔时间,单位ms（默认 100） send.msg.interval=100 #消息发送线程数量 （默认 3） send.msg.thread=3 #消息队列大小 （默认 500） msg.store.queue.size=500 #日志级别输出 (debug | error) logger.level=debug #自定义日志输出实现类 logger.implementation=io.growing.sdk.java.logger.GioLoggerImpl #运行模式，test:仅输出消息体，不发送消息，production: 发送消息 #run.mode=test run.mode=production # 设置代理, 如果不设置，默认为不使用代理 #proxy.host=127.0.0.1 #proxy.port=3128 # 设置代理认证用户密码, 如果不设置，默认为不使用用户验证 #proxy.user=demo #proxy.password=demo #http 连接超时时间 #connection.timeout=2000 #http 连接读取时间 #read.timeout=2000
