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();
|
// }
|
// }
|
}
|
