refactor: rename kafka sink module
This commit is contained in:
lingniu
@@ -0,0 +1,240 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import com.lingniu.ingest.api.event.AlarmPayload;
|
||||
import com.lingniu.ingest.api.event.LocationPayload;
|
||||
import com.lingniu.ingest.api.event.RawArchiveKeys;
|
||||
import com.lingniu.ingest.api.event.RealtimePayload;
|
||||
import com.lingniu.ingest.api.event.TelemetryFieldValue;
|
||||
import com.lingniu.ingest.api.event.TelemetrySnapshot;
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
import com.lingniu.ingest.api.event.VehicleEventTelemetrySnapshotMapper;
|
||||
import com.lingniu.ingest.facts.FactIds;
|
||||
import com.lingniu.ingest.facts.VehicleKey;
|
||||
import com.lingniu.ingest.sink.kafka.proto.ParseStatusProto;
|
||||
import com.lingniu.ingest.sink.kafka.proto.TelemetryField;
|
||||
import com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot.Builder;
|
||||
import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;
|
||||
|
||||
import java.security.MessageDigest;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.util.HexFormat;
|
||||
|
||||
/**
|
||||
* 领域事件 → Protobuf Envelope 的纯函数映射。通过 switch 模式匹配穷尽处理，
|
||||
* 新增 {@link VehicleEvent} 子类型时编译期就会提示补齐分支。
|
||||
*/
|
||||
public final class EnvelopeMapper {
|
||||
|
||||
private static final String SCHEMA_VERSION = "1.0";
|
||||
private final String nodeId;
|
||||
|
||||
public EnvelopeMapper(String nodeId) {
|
||||
this.nodeId = nodeId;
|
||||
}
|
||||
|
||||
public VehicleEnvelope toEnvelope(VehicleEvent event) {
|
||||
VehicleEnvelope.Builder b = VehicleEnvelope.newBuilder()
|
||||
.setSchemaVersion(SCHEMA_VERSION)
|
||||
.setEventId(event.eventId())
|
||||
.setTraceId(event.traceId() == null ? "" : event.traceId())
|
||||
.setVin(event.vin())
|
||||
.setSource(event.source().name())
|
||||
.setEventTimeMs(event.eventTime().toEpochMilli())
|
||||
.setIngestTimeMs(event.ingestTime().toEpochMilli())
|
||||
.setIngestNodeId(nodeId);
|
||||
if (event.metadata() != null) b.putAllMetadata(event.metadata());
|
||||
// telemetrySnapshot 是跨事件类型的统一字段视图，消费者可先读它，必要时再读具体 payload。
|
||||
VehicleEventTelemetrySnapshotMapper.toSnapshot(event)
|
||||
.map(EnvelopeMapper::buildTelemetrySnapshot)
|
||||
.ifPresent(b::setTelemetrySnapshot);
|
||||
|
||||
switch (event) {
|
||||
case VehicleEvent.Realtime r -> b.setRealtime(buildRealtime(r.payload()));
|
||||
case VehicleEvent.Location l -> b.setLocation(buildLocation(l.payload()));
|
||||
case VehicleEvent.Alarm a -> b.setAlarm(buildAlarm(a.payload()));
|
||||
case VehicleEvent.Login lg -> b.setLogin(
|
||||
com.lingniu.ingest.sink.kafka.proto.LoginPayload.newBuilder()
|
||||
.setIccid(nullToEmpty(lg.iccid()))
|
||||
.setProtocolVersion(nullToEmpty(lg.protocolVersion()))
|
||||
.build());
|
||||
case VehicleEvent.Logout ignored -> b.setLogout(
|
||||
com.lingniu.ingest.sink.kafka.proto.LogoutPayload.getDefaultInstance());
|
||||
case VehicleEvent.Heartbeat ignored -> b.setHeartbeat(
|
||||
com.lingniu.ingest.sink.kafka.proto.HeartbeatPayload.getDefaultInstance());
|
||||
case VehicleEvent.MediaMeta m -> b.setMediaMeta(
|
||||
com.lingniu.ingest.sink.kafka.proto.MediaMetaPayload.newBuilder()
|
||||
.setMediaId(nullToEmpty(m.mediaId()))
|
||||
.setMediaType(nullToEmpty(m.mediaType()))
|
||||
.setSizeBytes(m.sizeBytes())
|
||||
.setArchiveRef(nullToEmpty(m.archiveRef()))
|
||||
.build());
|
||||
case VehicleEvent.Passthrough p -> b.setPassthrough(
|
||||
com.lingniu.ingest.sink.kafka.proto.PassthroughPayload.newBuilder()
|
||||
.setPassthroughType(p.passthroughType())
|
||||
.setData(com.google.protobuf.ByteString.copyFrom(
|
||||
p.data() == null ? new byte[0] : p.data()))
|
||||
.build());
|
||||
case VehicleEvent.RawArchive ra -> {
|
||||
byte[] rawBytes = ra.rawBytes() == null ? new byte[0] : ra.rawBytes();
|
||||
int size = rawBytes.length;
|
||||
String key = rawArchiveKey(ra);
|
||||
String uri = rawArchiveUri(ra, key);
|
||||
String phone = metadataValue(ra, "phone");
|
||||
String vehicleKey = VehicleKey.derive(ra.source(), ra.vin(), phone, ra.eventId());
|
||||
String checksum = sha256(rawBytes);
|
||||
String frameId = FactIds.rawFrameId(
|
||||
ra.source(), vehicleKey, ra.command(), ra.infoType(), ra.ingestTime(), checksum);
|
||||
// RAW envelope 只携带 URI/size 引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。
|
||||
b.putMetadata(RawArchiveKeys.META_KEY, key);
|
||||
b.putMetadata(RawArchiveKeys.META_URI, uri);
|
||||
b.putMetadata(RawArchiveKeys.META_EVENT_ID, ra.eventId());
|
||||
b.setRawArchive(com.lingniu.ingest.sink.kafka.proto.RawArchiveRef.newBuilder()
|
||||
.setUri(uri)
|
||||
.setChecksum(checksum)
|
||||
.setSizeBytes(size)
|
||||
.setParsedJson(nullToEmpty(ra.parsedJson()))
|
||||
.build());
|
||||
b.setRawFrameFact(com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload.newBuilder()
|
||||
.setFrameId(frameId)
|
||||
.setVehicleKey(vehicleKey)
|
||||
.setVin(nullToEmpty(ra.vin()))
|
||||
.setPhone(phone)
|
||||
.setMessageId(ra.command())
|
||||
.setSubType(ra.infoType())
|
||||
.setRawUri(uri)
|
||||
.setChecksum(checksum)
|
||||
.setRawSizeBytes(size)
|
||||
.setParseStatus(parseStatus(ra))
|
||||
.setParseError(parseError(ra))
|
||||
.setPeer(metadataValue(ra, "peer"))
|
||||
.putAllMetadata(ra.metadata() == null ? java.util.Map.of() : ra.metadata())
|
||||
.build());
|
||||
}
|
||||
}
|
||||
return b.build();
|
||||
}
|
||||
|
||||
private static com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot buildTelemetrySnapshot(TelemetrySnapshot snapshot) {
|
||||
Builder b = com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot.newBuilder()
|
||||
.setEventType(snapshot.eventType())
|
||||
.setRawArchiveUri(snapshot.rawArchiveUri());
|
||||
for (TelemetryFieldValue field : snapshot.fields()) {
|
||||
b.addFields(TelemetryField.newBuilder()
|
||||
.setKey(field.key())
|
||||
.setValueType(field.valueType().name())
|
||||
.setValue(field.value())
|
||||
.setUnit(field.unit())
|
||||
.setQuality(field.quality().name())
|
||||
.setSourcePath(field.sourcePath())
|
||||
.build());
|
||||
}
|
||||
return b.build();
|
||||
}
|
||||
|
||||
private static com.lingniu.ingest.sink.kafka.proto.RealtimePayload buildRealtime(RealtimePayload p) {
|
||||
var b = com.lingniu.ingest.sink.kafka.proto.RealtimePayload.newBuilder();
|
||||
if (p.speedKmh() != null) b.setSpeedKmh(p.speedKmh());
|
||||
if (p.totalMileageKm() != null) b.setTotalMileageKm(p.totalMileageKm());
|
||||
if (p.batterySoc() != null) b.setBatterySoc(p.batterySoc());
|
||||
if (p.batteryVoltageV() != null) b.setBatteryVoltageV(p.batteryVoltageV());
|
||||
if (p.batteryCurrentA() != null) b.setBatteryCurrentA(p.batteryCurrentA());
|
||||
if (p.fcVoltageV() != null) b.setFcVoltageV(p.fcVoltageV());
|
||||
if (p.fcCurrentA() != null) b.setFcCurrentA(p.fcCurrentA());
|
||||
if (p.fcTempC() != null) b.setFcTempC(p.fcTempC());
|
||||
if (p.hydrogenRemainingKg() != null) b.setHydrogenRemainingKg(p.hydrogenRemainingKg());
|
||||
if (p.hydrogenHighPressureMpa() != null) b.setHydrogenHighPressureMpa(p.hydrogenHighPressureMpa());
|
||||
if (p.hydrogenLowPressureMpa() != null) b.setHydrogenLowPressureMpa(p.hydrogenLowPressureMpa());
|
||||
if (p.vehicleState() != null) b.setVehicleState(p.vehicleState().name());
|
||||
if (p.chargingState() != null) b.setChargingState(p.chargingState().name());
|
||||
if (p.runningMode() != null) b.setRunningMode(p.runningMode().name());
|
||||
if (p.gearLevel() != null) b.setGearLevel(p.gearLevel());
|
||||
if (p.acceleratorPedal() != null) b.setAcceleratorPedal(p.acceleratorPedal());
|
||||
if (p.brakePedal() != null) b.setBrakePedal(p.brakePedal());
|
||||
if (p.longitude() != null) b.setLongitude(p.longitude());
|
||||
if (p.latitude() != null) b.setLatitude(p.latitude());
|
||||
if (p.altitudeM() != null) b.setAltitudeM(p.altitudeM());
|
||||
if (p.directionDeg() != null) b.setDirectionDeg(p.directionDeg());
|
||||
if (p.ambientTempC() != null) b.setAmbientTempC(p.ambientTempC());
|
||||
if (p.coolantTempC() != null) b.setCoolantTempC(p.coolantTempC());
|
||||
return b.build();
|
||||
}
|
||||
|
||||
private static com.lingniu.ingest.sink.kafka.proto.LocationPayload buildLocation(LocationPayload p) {
|
||||
return com.lingniu.ingest.sink.kafka.proto.LocationPayload.newBuilder()
|
||||
.setLongitude(p.longitude())
|
||||
.setLatitude(p.latitude())
|
||||
.setAltitudeM(p.altitudeM())
|
||||
.setSpeedKmh(p.speedKmh())
|
||||
.setDirectionDeg(p.directionDeg())
|
||||
.setAlarmFlag(p.alarmFlag())
|
||||
.setStatusFlag(p.statusFlag())
|
||||
.build();
|
||||
}
|
||||
|
||||
private static com.lingniu.ingest.sink.kafka.proto.AlarmPayload buildAlarm(AlarmPayload p) {
|
||||
var b = com.lingniu.ingest.sink.kafka.proto.AlarmPayload.newBuilder()
|
||||
.setLevel(p.level().name())
|
||||
.setAlarmTypeCode(p.alarmTypeCode())
|
||||
.setAlarmTypeName(nullToEmpty(p.alarmTypeName()));
|
||||
if (p.faultCodes() != null) b.addAllFaultCodes(p.faultCodes());
|
||||
if (p.activeBits() != null) b.addAllActiveBits(p.activeBits());
|
||||
if (p.longitude() != null) b.setLongitude(p.longitude());
|
||||
if (p.latitude() != null) b.setLatitude(p.latitude());
|
||||
return b.build();
|
||||
}
|
||||
|
||||
private static String nullToEmpty(String s) {
|
||||
return s == null ? "" : s;
|
||||
}
|
||||
|
||||
private static String rawArchiveKey(VehicleEvent.RawArchive raw) {
|
||||
String key = raw.metadata() == null ? "" : raw.metadata().getOrDefault(RawArchiveKeys.META_KEY, "");
|
||||
return key == null || key.isBlank() ? RawArchiveKeys.key(raw) : key;
|
||||
}
|
||||
|
||||
private static String rawArchiveUri(VehicleEvent.RawArchive raw, String key) {
|
||||
String uri = raw.metadata() == null ? "" : raw.metadata().getOrDefault(RawArchiveKeys.META_URI, "");
|
||||
return uri == null || uri.isBlank() ? RawArchiveKeys.logicalUri(key) : uri;
|
||||
}
|
||||
|
||||
private static String metadataValue(VehicleEvent event, String key) {
|
||||
if (event.metadata() == null) {
|
||||
return "";
|
||||
}
|
||||
return nullToEmpty(event.metadata().get(key));
|
||||
}
|
||||
|
||||
private static ParseStatusProto parseStatus(VehicleEvent.RawArchive raw) {
|
||||
if (hasTruthyMetadata(raw, "frameError")
|
||||
|| hasTruthyMetadata(raw, "parseError")
|
||||
|| hasTruthyMetadata(raw, "processingError")) {
|
||||
return ParseStatusProto.PARSE_STATUS_FAILED;
|
||||
}
|
||||
return ParseStatusProto.PARSE_STATUS_NOT_PARSED;
|
||||
}
|
||||
|
||||
private static String parseError(VehicleEvent.RawArchive raw) {
|
||||
String frameError = metadataValue(raw, "frameErrorMessage");
|
||||
if (!frameError.isBlank()) {
|
||||
return frameError;
|
||||
}
|
||||
String parseError = metadataValue(raw, "parseErrorMessage");
|
||||
if (!parseError.isBlank()) {
|
||||
return parseError;
|
||||
}
|
||||
return metadataValue(raw, "processingErrorMessage");
|
||||
}
|
||||
|
||||
private static boolean hasTruthyMetadata(VehicleEvent event, String key) {
|
||||
return Boolean.parseBoolean(metadataValue(event, key));
|
||||
}
|
||||
|
||||
private static String sha256(byte[] bytes) {
|
||||
try {
|
||||
MessageDigest digest = MessageDigest.getInstance("SHA-256");
|
||||
return "sha256:" + HexFormat.of().formatHex(digest.digest(bytes == null ? new byte[0] : bytes));
|
||||
} catch (NoSuchAlgorithmException e) {
|
||||
throw new IllegalStateException("SHA-256 is unavailable", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,142 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import org.apache.kafka.clients.consumer.ConsumerConfig;
|
||||
import org.apache.kafka.clients.consumer.KafkaConsumer;
|
||||
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
|
||||
import org.apache.kafka.common.serialization.StringDeserializer;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
import java.util.function.Function;
|
||||
|
||||
public final class KafkaEnvelopeConsumerFactory {
|
||||
|
||||
private final Function<Properties, org.apache.kafka.clients.consumer.Consumer<String, byte[]>> consumerFactory;
|
||||
|
||||
public KafkaEnvelopeConsumerFactory() {
|
||||
this(KafkaConsumer::new);
|
||||
}
|
||||
|
||||
KafkaEnvelopeConsumerFactory(
|
||||
Function<Properties, org.apache.kafka.clients.consumer.Consumer<String, byte[]>> consumerFactory) {
|
||||
this.consumerFactory = consumerFactory;
|
||||
}
|
||||
|
||||
public List<KafkaEnvelopeConsumerWorker> createWorkers(Map<String, EnvelopeConsumerProcessor> processors,
|
||||
KafkaSinkProperties props) {
|
||||
Map<String, KafkaSinkProperties.Binding> bindings = effectiveBindings(props);
|
||||
List<KafkaEnvelopeConsumerWorker> workers = new ArrayList<>();
|
||||
for (Map.Entry<String, KafkaSinkProperties.Binding> entry : bindings.entrySet()) {
|
||||
String processorBeanName = entry.getKey();
|
||||
// binding 的 key 必须和 Spring Bean 名一致；这样配置只声明 topic/group，
|
||||
// 实际处理逻辑仍由各业务模块自己的 EnvelopeConsumerProcessor 承接。
|
||||
EnvelopeConsumerProcessor processor = processors.get(processorBeanName);
|
||||
KafkaSinkProperties.Binding binding = entry.getValue();
|
||||
if (processor == null || binding == null || !binding.isEnabled()) {
|
||||
continue;
|
||||
}
|
||||
List<String> topics = cleanTopics(binding.getTopics());
|
||||
if (topics.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
int concurrency = Math.max(1, props.getConsumer().getConcurrency());
|
||||
for (int workerIndex = 0; workerIndex < concurrency; workerIndex++) {
|
||||
KafkaEnvelopeConsumerWorker worker = new KafkaEnvelopeConsumerWorker(
|
||||
consumerFactory.apply(consumerProperties(props, binding, processorBeanName,
|
||||
workerIndex, concurrency)),
|
||||
topicProcessors(topics, processor));
|
||||
worker.subscribe(topics);
|
||||
workers.add(worker);
|
||||
}
|
||||
}
|
||||
return workers;
|
||||
}
|
||||
|
||||
private Map<String, KafkaSinkProperties.Binding> effectiveBindings(KafkaSinkProperties props) {
|
||||
Map<String, KafkaSinkProperties.Binding> configured = props.getConsumer().getBindings();
|
||||
if (configured != null && !configured.isEmpty()) {
|
||||
return configured;
|
||||
}
|
||||
// 默认绑定仅给未显式配置 bindings 的轻量运行时兜底。
|
||||
// 生产 history app 会显式绑定各协议 event/raw topic，并写入 TDengine raw_frames/locations。
|
||||
KafkaSinkProperties.Topics topics = props.getTopics();
|
||||
Map<String, KafkaSinkProperties.Binding> defaults = new LinkedHashMap<>();
|
||||
defaults.put("eventHistoryEnvelopeConsumerProcessor", binding(
|
||||
"vehicle-event-history",
|
||||
topics.getRealtime(), topics.getLocation(), topics.getAlarm(), topics.getSession(), topics.getMediaMeta()));
|
||||
defaults.put("vehicleStateEnvelopeConsumerProcessor", binding(
|
||||
"vehicle-state",
|
||||
topics.getRealtime(), topics.getLocation(), topics.getAlarm()));
|
||||
defaults.put("vehicleStatEnvelopeConsumerProcessor", binding(
|
||||
"vehicle-stat",
|
||||
topics.getRealtime(), topics.getLocation()));
|
||||
return defaults;
|
||||
}
|
||||
|
||||
private KafkaSinkProperties.Binding binding(String groupId, String... topics) {
|
||||
KafkaSinkProperties.Binding binding = new KafkaSinkProperties.Binding();
|
||||
binding.setGroupId(groupId);
|
||||
binding.setTopics(List.of(topics));
|
||||
return binding;
|
||||
}
|
||||
|
||||
private Map<String, EnvelopeConsumerProcessor> topicProcessors(List<String> topics, EnvelopeConsumerProcessor processor) {
|
||||
Map<String, EnvelopeConsumerProcessor> byTopic = new LinkedHashMap<>();
|
||||
for (String topic : topics) {
|
||||
byTopic.put(topic, processor);
|
||||
}
|
||||
return byTopic;
|
||||
}
|
||||
|
||||
private List<String> cleanTopics(List<String> topics) {
|
||||
if (topics == null || topics.isEmpty()) {
|
||||
return List.of();
|
||||
}
|
||||
LinkedHashSet<String> clean = new LinkedHashSet<>();
|
||||
for (String topic : topics) {
|
||||
if (topic != null && !topic.isBlank()) {
|
||||
clean.add(topic);
|
||||
}
|
||||
}
|
||||
return List.copyOf(clean);
|
||||
}
|
||||
|
||||
private Properties consumerProperties(KafkaSinkProperties props,
|
||||
KafkaSinkProperties.Binding binding,
|
||||
String processorBeanName,
|
||||
int workerIndex,
|
||||
int concurrency) {
|
||||
KafkaSinkProperties.Consumer consumer = props.getConsumer();
|
||||
Properties p = new Properties();
|
||||
p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getBootstrapServers());
|
||||
p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
|
||||
p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
|
||||
p.put(ConsumerConfig.GROUP_ID_CONFIG, groupId(binding, processorBeanName));
|
||||
p.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId(consumer.getClientIdPrefix(), processorBeanName,
|
||||
workerIndex, concurrency));
|
||||
// 手动提交 offset：只有本批次至少有一条记录被处理器接收后才 commit，
|
||||
// 避免轮询到空批次或未绑定 topic 时推进消费位点。
|
||||
p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
|
||||
p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, consumer.getAutoOffsetReset());
|
||||
p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, consumer.getMaxPollRecords());
|
||||
p.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, consumer.getMaxPollIntervalMillis());
|
||||
return p;
|
||||
}
|
||||
|
||||
private String clientId(String prefix, String processorBeanName, int workerIndex, int concurrency) {
|
||||
String base = prefix + "-" + processorBeanName;
|
||||
return concurrency <= 1 ? base : base + "-" + workerIndex;
|
||||
}
|
||||
|
||||
private String groupId(KafkaSinkProperties.Binding binding, String processorBeanName) {
|
||||
if (binding.getGroupId() != null && !binding.getGroupId().isBlank()) {
|
||||
return binding.getGroupId();
|
||||
}
|
||||
return processorBeanName.replace("EnvelopeConsumerProcessor", "");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,182 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.context.SmartLifecycle;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
public final class KafkaEnvelopeConsumerRunner implements SmartLifecycle, AutoCloseable {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(KafkaEnvelopeConsumerRunner.class);
|
||||
|
||||
private final Supplier<List<KafkaEnvelopeConsumerWorker>> workersSupplier;
|
||||
private final Duration pollTimeout;
|
||||
private final Duration loopBackoff;
|
||||
private final boolean autoStartup;
|
||||
private final AtomicBoolean running = new AtomicBoolean(false);
|
||||
private final AtomicBoolean closed = new AtomicBoolean(false);
|
||||
private volatile List<KafkaEnvelopeConsumerWorker> workers;
|
||||
private ExecutorService executor;
|
||||
|
||||
public KafkaEnvelopeConsumerRunner(List<KafkaEnvelopeConsumerWorker> workers,
|
||||
Duration pollTimeout,
|
||||
Duration loopBackoff,
|
||||
boolean autoStartup) {
|
||||
if (workers == null || workers.isEmpty()) {
|
||||
throw new IllegalArgumentException("workers must not be empty");
|
||||
}
|
||||
this.workersSupplier = () -> List.copyOf(workers);
|
||||
this.workers = List.copyOf(workers);
|
||||
this.pollTimeout = pollTimeout == null ? Duration.ofSeconds(1) : pollTimeout;
|
||||
this.loopBackoff = loopBackoff == null ? Duration.ofSeconds(1) : loopBackoff;
|
||||
this.autoStartup = autoStartup;
|
||||
}
|
||||
|
||||
public KafkaEnvelopeConsumerRunner(Supplier<List<KafkaEnvelopeConsumerWorker>> workersSupplier,
|
||||
Duration pollTimeout,
|
||||
Duration loopBackoff,
|
||||
boolean autoStartup) {
|
||||
if (workersSupplier == null) {
|
||||
throw new IllegalArgumentException("workersSupplier must not be null");
|
||||
}
|
||||
this.workersSupplier = workersSupplier;
|
||||
this.pollTimeout = pollTimeout == null ? Duration.ofSeconds(1) : pollTimeout;
|
||||
this.loopBackoff = loopBackoff == null ? Duration.ofSeconds(1) : loopBackoff;
|
||||
this.autoStartup = autoStartup;
|
||||
}
|
||||
|
||||
public List<KafkaEnvelopeConsumerWorker> workers() {
|
||||
return workersOrCreate();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
if (!running.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
List<KafkaEnvelopeConsumerWorker> activeWorkers = workersOrCreate();
|
||||
// 每个 worker 一条后台线程，避免某个处理器阻塞时拖慢其他消费组。
|
||||
executor = Executors.newFixedThreadPool(activeWorkers.size(), r -> {
|
||||
Thread thread = new Thread(r, "kafka-envelope-consumer");
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
});
|
||||
for (KafkaEnvelopeConsumerWorker worker : activeWorkers) {
|
||||
executor.submit(() -> pollLoop(worker));
|
||||
}
|
||||
}
|
||||
|
||||
private List<KafkaEnvelopeConsumerWorker> workersOrCreate() {
|
||||
List<KafkaEnvelopeConsumerWorker> current = workers;
|
||||
if (current != null) {
|
||||
return current;
|
||||
}
|
||||
synchronized (this) {
|
||||
if (workers == null) {
|
||||
List<KafkaEnvelopeConsumerWorker> created = workersSupplier.get();
|
||||
if (created == null || created.isEmpty()) {
|
||||
throw new IllegalStateException("no kafka envelope consumer workers created; check consumer bindings");
|
||||
}
|
||||
workers = List.copyOf(created);
|
||||
}
|
||||
return workers;
|
||||
}
|
||||
}
|
||||
|
||||
private void pollLoop(KafkaEnvelopeConsumerWorker worker) {
|
||||
while (running.get()) {
|
||||
try {
|
||||
worker.pollOnce(pollTimeout);
|
||||
} catch (RuntimeException ex) {
|
||||
if (!running.get()) {
|
||||
break;
|
||||
}
|
||||
// Kafka/处理器异常不让 Spring 生命周期退出；退避后继续消费，
|
||||
// 具体坏消息由 EnvelopeConsumerProcessor 写入 DLQ。
|
||||
log.warn("Kafka envelope consumer poll failed; the worker will retry after backoff", ex);
|
||||
sleepBackoff();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void sleepBackoff() {
|
||||
try {
|
||||
Thread.sleep(loopBackoff.toMillis());
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
if (!running.compareAndSet(true, false)) {
|
||||
return;
|
||||
}
|
||||
wakeupWorkers();
|
||||
if (executor != null) {
|
||||
executor.shutdown();
|
||||
try {
|
||||
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
|
||||
executor.shutdownNow();
|
||||
executor.awaitTermination(5, TimeUnit.SECONDS);
|
||||
}
|
||||
} catch (InterruptedException ex) {
|
||||
executor.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
closeWorkers();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop(Runnable callback) {
|
||||
stop();
|
||||
callback.run();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return running.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return autoStartup;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
stop();
|
||||
closeWorkers();
|
||||
}
|
||||
|
||||
private void closeWorkers() {
|
||||
if (!closed.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
List<KafkaEnvelopeConsumerWorker> current = workers;
|
||||
if (current == null) {
|
||||
return;
|
||||
}
|
||||
for (KafkaEnvelopeConsumerWorker worker : current) {
|
||||
worker.close();
|
||||
}
|
||||
}
|
||||
|
||||
private void wakeupWorkers() {
|
||||
List<KafkaEnvelopeConsumerWorker> current = workers;
|
||||
if (current == null) {
|
||||
return;
|
||||
}
|
||||
for (KafkaEnvelopeConsumerWorker worker : current) {
|
||||
worker.wakeup();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.Consumer;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecords;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
public final class KafkaEnvelopeConsumerWorker implements AutoCloseable {
|
||||
|
||||
private final Consumer<String, byte[]> consumer;
|
||||
private final Map<String, EnvelopeConsumerProcessor> processorsByTopic;
|
||||
|
||||
public KafkaEnvelopeConsumerWorker(Consumer<String, byte[]> consumer,
|
||||
Map<String, EnvelopeConsumerProcessor> processorsByTopic) {
|
||||
if (consumer == null) {
|
||||
throw new IllegalArgumentException("consumer must not be null");
|
||||
}
|
||||
if (processorsByTopic == null || processorsByTopic.isEmpty()) {
|
||||
throw new IllegalArgumentException("processorsByTopic must not be empty");
|
||||
}
|
||||
this.consumer = consumer;
|
||||
this.processorsByTopic = Map.copyOf(processorsByTopic);
|
||||
}
|
||||
|
||||
public void subscribe(Collection<String> topics) {
|
||||
consumer.subscribe(topics);
|
||||
}
|
||||
|
||||
public int pollOnce(Duration timeout) {
|
||||
ConsumerRecords<String, byte[]> records = consumer.poll(timeout == null ? Duration.ZERO : timeout);
|
||||
Map<EnvelopeConsumerProcessor, List<EnvelopeConsumerRecord>> byProcessor = new LinkedHashMap<>();
|
||||
int processed = 0;
|
||||
for (ConsumerRecord<String, byte[]> record : records) {
|
||||
EnvelopeConsumerProcessor processor = processorsByTopic.get(record.topic());
|
||||
if (processor == null) {
|
||||
// worker 可能订阅多个 topic；没有显式绑定处理器的 topic 不参与提交语义。
|
||||
continue;
|
||||
}
|
||||
byProcessor.computeIfAbsent(processor, ignored -> new ArrayList<>()).add(new EnvelopeConsumerRecord(
|
||||
record.topic(),
|
||||
record.partition(),
|
||||
record.offset(),
|
||||
record.key(),
|
||||
record.value()));
|
||||
processed++;
|
||||
}
|
||||
for (Map.Entry<EnvelopeConsumerProcessor, List<EnvelopeConsumerRecord>> entry : byProcessor.entrySet()) {
|
||||
// EnvelopeConsumerProcessor 内部会把解析或业务错误转成 DLQ 记录，
|
||||
// 这里保持 Kafka worker 的职责单一：轮询、分发、成功后提交 offset。
|
||||
entry.getKey().processBatch(entry.getValue());
|
||||
}
|
||||
if (processed > 0) {
|
||||
// commitSync 放在批次末尾，保证同一个 poll 批次内的消息按 Kafka offset 一起确认。
|
||||
consumer.commitSync();
|
||||
}
|
||||
return processed;
|
||||
}
|
||||
|
||||
public void wakeup() {
|
||||
consumer.wakeup();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
consumer.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterRecord;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.Producer;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
public final class KafkaEnvelopeDeadLetterSink implements EnvelopeDeadLetterSink {
|
||||
|
||||
private final Producer<String, byte[]> producer;
|
||||
private final String topic;
|
||||
|
||||
public KafkaEnvelopeDeadLetterSink(KafkaProducer<String, byte[]> producer, String topic) {
|
||||
this((Producer<String, byte[]>) producer, topic);
|
||||
}
|
||||
|
||||
KafkaEnvelopeDeadLetterSink(Producer<String, byte[]> producer, String topic) {
|
||||
if (producer == null) {
|
||||
throw new IllegalArgumentException("producer must not be null");
|
||||
}
|
||||
if (topic == null || topic.isBlank()) {
|
||||
throw new IllegalArgumentException("topic must not be blank");
|
||||
}
|
||||
this.producer = producer;
|
||||
this.topic = topic;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void publish(EnvelopeDeadLetterRecord record) {
|
||||
if (record == null) {
|
||||
throw new IllegalArgumentException("record must not be null");
|
||||
}
|
||||
ProducerRecord<String, byte[]> out = new ProducerRecord<>(topic, record.key(), record.payload());
|
||||
// DLQ payload 保持原始 Kafka value；定位信息全部放 header，便于后续重放或人工排查。
|
||||
header(out, "dlq-service", record.service());
|
||||
header(out, "dlq-source-topic", record.topic());
|
||||
header(out, "dlq-source-partition", Integer.toString(record.partition()));
|
||||
header(out, "dlq-source-offset", Long.toString(record.offset()));
|
||||
header(out, "dlq-status", record.status().name());
|
||||
header(out, "dlq-event-id", record.eventId());
|
||||
header(out, "dlq-vin", record.vin());
|
||||
header(out, "dlq-message", record.message());
|
||||
header(out, "dlq-created-at", record.createdAt().toString());
|
||||
producer.send(out);
|
||||
}
|
||||
|
||||
private static void header(ProducerRecord<String, byte[]> record, String key, String value) {
|
||||
record.headers().add(key, value.getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
import com.lingniu.ingest.api.sink.EventSink;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
/**
|
||||
* Kafka Sink：序列化为 Protobuf Envelope，按 vin 分区，异步发送。
|
||||
*
|
||||
* <p>失败处理：Resilience4j 熔断；熔断开启时事件转投 DLQ topic。
|
||||
*/
|
||||
public final class KafkaEventSink implements EventSink, AutoCloseable {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(KafkaEventSink.class);
|
||||
|
||||
private final KafkaProducer<String, byte[]> producer;
|
||||
private final EnvelopeMapper mapper;
|
||||
private final TopicRouter router;
|
||||
private final String dlqTopic;
|
||||
private final CircuitBreaker breaker;
|
||||
|
||||
public KafkaEventSink(KafkaProducer<String, byte[]> producer,
|
||||
EnvelopeMapper mapper,
|
||||
TopicRouter router,
|
||||
String dlqTopic,
|
||||
CircuitBreaker breaker) {
|
||||
this.producer = producer;
|
||||
this.mapper = mapper;
|
||||
this.router = router;
|
||||
this.dlqTopic = dlqTopic;
|
||||
this.breaker = breaker;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return "kafka";
|
||||
}
|
||||
|
||||
/**
|
||||
* Split-service mode uses Kafka for both normalized events and raw archive records.
|
||||
*/
|
||||
@Override
|
||||
public boolean accepts(VehicleEvent event) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletableFuture<Void> publish(VehicleEvent event) {
|
||||
CompletableFuture<Void> cf = new CompletableFuture<>();
|
||||
byte[] payload;
|
||||
try {
|
||||
payload = mapper.toEnvelope(event).toByteArray();
|
||||
} catch (Exception e) {
|
||||
log.error("envelope build failed eventId={} vin={}", event.eventId(), event.vin(), e);
|
||||
cf.completeExceptionally(e);
|
||||
return cf;
|
||||
}
|
||||
|
||||
String topic = breaker.tryAcquirePermission() ? router.route(event) : dlqTopic;
|
||||
ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, partitionKey(event), payload);
|
||||
record.headers().add("event-id", event.eventId().getBytes());
|
||||
record.headers().add("trace-id", event.traceId() == null ? new byte[0] : event.traceId().getBytes());
|
||||
record.headers().add("source", event.source().name().getBytes());
|
||||
|
||||
producer.send(record, (metadata, ex) -> {
|
||||
if (ex != null) {
|
||||
breaker.onError(0, java.util.concurrent.TimeUnit.MILLISECONDS, ex);
|
||||
cf.completeExceptionally(ex);
|
||||
} else {
|
||||
breaker.onSuccess(0, java.util.concurrent.TimeUnit.MILLISECONDS);
|
||||
cf.complete(null);
|
||||
}
|
||||
});
|
||||
return cf;
|
||||
}
|
||||
|
||||
private static String partitionKey(VehicleEvent event) {
|
||||
String vin = event.vin();
|
||||
if (isKnown(vin)) {
|
||||
return vin;
|
||||
}
|
||||
String vehicleKey = metadata(event, "vehicleKey");
|
||||
if (!isKnown(vehicleKey)) {
|
||||
vehicleKey = metadata(event, "vehicle_key");
|
||||
}
|
||||
if (isKnown(vehicleKey)) {
|
||||
return "vehicleKey:" + vehicleKey;
|
||||
}
|
||||
String phone = metadata(event, "phone");
|
||||
if (isKnown(phone)) {
|
||||
return "phone:" + phone;
|
||||
}
|
||||
return vin == null || vin.isBlank() ? "unknown" : vin;
|
||||
}
|
||||
|
||||
private static String metadata(VehicleEvent event, String key) {
|
||||
if (event.metadata() == null) {
|
||||
return "";
|
||||
}
|
||||
return event.metadata().getOrDefault(key, "").trim();
|
||||
}
|
||||
|
||||
private static boolean isKnown(String value) {
|
||||
return value != null && !value.isBlank() && !"unknown".equalsIgnoreCase(value.trim());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
producer.flush();
|
||||
producer.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.ProducerConfig;
|
||||
import org.apache.kafka.common.serialization.ByteArraySerializer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Properties;
|
||||
|
||||
/**
|
||||
* Kafka Sink 自动装配。
|
||||
*
|
||||
* <p>{@code lingniu.ingest.sink.kafka.enabled=false} 时本模块完全不装配；Producer 和 Consumer
|
||||
* 是两个独立开关，消费端还需要开启 {@code lingniu.ingest.sink.kafka.consumer.enabled=true}
|
||||
* 并配置 bindings。
|
||||
*/
|
||||
@AutoConfiguration
|
||||
@EnableConfigurationProperties(KafkaSinkProperties.class)
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.kafka", name = "enabled", havingValue = "true", matchIfMissing = true)
|
||||
public class KafkaSinkAutoConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public EnvelopeMapper envelopeMapper(KafkaSinkProperties props) {
|
||||
return new EnvelopeMapper(props.getNodeId());
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public TopicRouter topicRouter(KafkaSinkProperties props) {
|
||||
return new TopicRouter(props.getTopics());
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public CircuitBreaker kafkaSinkCircuitBreaker() {
|
||||
return CircuitBreaker.of("kafka-sink", CircuitBreakerConfig.custom()
|
||||
.slidingWindowSize(100)
|
||||
.failureRateThreshold(50)
|
||||
.waitDurationInOpenState(Duration.ofSeconds(10))
|
||||
.permittedNumberOfCallsInHalfOpenState(10)
|
||||
.build());
|
||||
}
|
||||
|
||||
@Bean(destroyMethod = "close")
|
||||
@ConditionalOnMissingBean
|
||||
public KafkaProducer<String, byte[]> kafkaProducer(KafkaSinkProperties props) {
|
||||
Properties p = new Properties();
|
||||
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getBootstrapServers());
|
||||
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
|
||||
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
|
||||
p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, props.getCompressionType());
|
||||
p.put(ProducerConfig.LINGER_MS_CONFIG, props.getLingerMs());
|
||||
p.put(ProducerConfig.BATCH_SIZE_CONFIG, props.getBatchSize());
|
||||
p.put(ProducerConfig.ACKS_CONFIG, props.getAcks());
|
||||
p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, props.isEnableIdempotence());
|
||||
return new KafkaProducer<>(p);
|
||||
}
|
||||
|
||||
@Bean(destroyMethod = "close")
|
||||
@ConditionalOnMissingBean
|
||||
public KafkaEventSink kafkaEventSink(KafkaProducer<String, byte[]> producer,
|
||||
EnvelopeMapper mapper,
|
||||
TopicRouter router,
|
||||
KafkaSinkProperties props,
|
||||
CircuitBreaker breaker) {
|
||||
// KafkaEventSink 是 EventBus 的生产端出口；它不会启动任何 Kafka 消费线程。
|
||||
return new KafkaEventSink(producer, mapper, router, props.getTopics().getDlq(), breaker);
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public KafkaEnvelopeDeadLetterSink kafkaEnvelopeDeadLetterSink(KafkaProducer<String, byte[]> producer,
|
||||
KafkaSinkProperties props) {
|
||||
return new KafkaEnvelopeDeadLetterSink(producer, props.getTopics().getDlq());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import org.springframework.beans.factory.ListableBeanFactory;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@AutoConfiguration(after = KafkaSinkAutoConfiguration.class)
|
||||
@AutoConfigureAfter(name = {
|
||||
"com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration",
|
||||
"com.lingniu.ingest.vehiclestate.config.VehicleStateAutoConfiguration",
|
||||
"com.lingniu.ingest.vehiclestat.config.VehicleStatAutoConfiguration"
|
||||
})
|
||||
@EnableConfigurationProperties(KafkaSinkProperties.class)
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.kafka", name = "enabled", havingValue = "true", matchIfMissing = true)
|
||||
public class KafkaSinkConsumerAutoConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public KafkaEnvelopeConsumerFactory kafkaEnvelopeConsumerFactory() {
|
||||
return new KafkaEnvelopeConsumerFactory();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.kafka.consumer", name = "enabled", havingValue = "true")
|
||||
public KafkaEnvelopeConsumerRunner kafkaEnvelopeConsumerRunner(ListableBeanFactory beanFactory,
|
||||
KafkaEnvelopeConsumerFactory consumerFactory,
|
||||
KafkaSinkProperties props) {
|
||||
return new KafkaEnvelopeConsumerRunner(
|
||||
() -> createWorkers(beanFactory, consumerFactory, props),
|
||||
Duration.ofMillis(props.getConsumer().getPollTimeoutMillis()),
|
||||
Duration.ofMillis(props.getConsumer().getLoopBackoffMillis()),
|
||||
props.getConsumer().isAutoStartup());
|
||||
}
|
||||
|
||||
private List<KafkaEnvelopeConsumerWorker> createWorkers(ListableBeanFactory beanFactory,
|
||||
KafkaEnvelopeConsumerFactory consumerFactory,
|
||||
KafkaSinkProperties props) {
|
||||
Map<String, EnvelopeConsumerProcessor> processors = beanFactory.getBeansOfType(EnvelopeConsumerProcessor.class);
|
||||
return consumerFactory.createWorkers(processors, props);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,139 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@ConfigurationProperties(prefix = "lingniu.ingest.sink.kafka")
|
||||
public class KafkaSinkProperties {
|
||||
|
||||
/**
|
||||
* Kafka Sink 总开关。默认 {@code true}。
|
||||
* 设为 {@code false} 时 {@link KafkaSinkAutoConfiguration} 完全不装配任何 Bean（Kafka Producer、
|
||||
* EnvelopeMapper、TopicRouter、KafkaEventSink 都不会创建），ingest-core 的 DisruptorEventBus
|
||||
* 仍然正常运行但没有外部 sink（事件落地到 sink-archive 或 Noop 吞掉）。
|
||||
*/
|
||||
private boolean enabled = true;
|
||||
|
||||
/** Kafka bootstrap servers；生产环境应通过环境变量覆盖，不建议使用默认开发地址。 */
|
||||
private String bootstrapServers = "114.55.58.251:9092";
|
||||
private String compressionType = "zstd";
|
||||
private int lingerMs = 20;
|
||||
private int batchSize = 65536;
|
||||
private String acks = "all";
|
||||
private boolean enableIdempotence = true;
|
||||
/** 写入 envelope 的节点标识，进入 Protobuf 字段 ingest_node_id，便于追踪多实例来源。 */
|
||||
private String nodeId = "ingest-local";
|
||||
private Topics topics = new Topics();
|
||||
private Consumer consumer = new Consumer();
|
||||
|
||||
public boolean isEnabled() { return enabled; }
|
||||
public void setEnabled(boolean enabled) { this.enabled = enabled; }
|
||||
public String getBootstrapServers() { return bootstrapServers; }
|
||||
public void setBootstrapServers(String bootstrapServers) { this.bootstrapServers = bootstrapServers; }
|
||||
public String getCompressionType() { return compressionType; }
|
||||
public void setCompressionType(String compressionType) { this.compressionType = compressionType; }
|
||||
public int getLingerMs() { return lingerMs; }
|
||||
public void setLingerMs(int lingerMs) { this.lingerMs = lingerMs; }
|
||||
public int getBatchSize() { return batchSize; }
|
||||
public void setBatchSize(int batchSize) { this.batchSize = batchSize; }
|
||||
public String getAcks() { return acks; }
|
||||
public void setAcks(String acks) { this.acks = acks; }
|
||||
public boolean isEnableIdempotence() { return enableIdempotence; }
|
||||
public void setEnableIdempotence(boolean enableIdempotence) { this.enableIdempotence = enableIdempotence; }
|
||||
public String getNodeId() { return nodeId; }
|
||||
public void setNodeId(String nodeId) { this.nodeId = nodeId; }
|
||||
public Topics getTopics() { return topics; }
|
||||
public void setTopics(Topics topics) { this.topics = topics; }
|
||||
public Consumer getConsumer() { return consumer; }
|
||||
public void setConsumer(Consumer consumer) { this.consumer = consumer; }
|
||||
|
||||
public static class Topics {
|
||||
/** 实时遥测事件 topic；GB32960 RAW-only 架构下可逐步弱化该 topic。 */
|
||||
private String realtime = "vehicle.event.gb32960.v1";
|
||||
/** 位置事件 topic；通常可由 realtime/RAW 派生。 */
|
||||
private String location = "vehicle.event.gb32960.v1";
|
||||
private String alarm = "vehicle.event.gb32960.v1";
|
||||
private String session = "vehicle.event.gb32960.v1";
|
||||
private String mediaMeta = "vehicle.media.meta.v1";
|
||||
/** RAW 归档引用 topic，payload 应携带 archive URI/size，不建议携带完整 raw bytes。 */
|
||||
private String rawArchive = "vehicle.raw.gb32960.v1";
|
||||
/** producer 熔断或 consumer 处理失败时的死信 topic。 */
|
||||
private String dlq = "vehicle.dlq.gb32960.v1";
|
||||
|
||||
public String getRealtime() { return realtime; }
|
||||
public void setRealtime(String realtime) { this.realtime = realtime; }
|
||||
public String getLocation() { return location; }
|
||||
public void setLocation(String location) { this.location = location; }
|
||||
public String getAlarm() { return alarm; }
|
||||
public void setAlarm(String alarm) { this.alarm = alarm; }
|
||||
public String getSession() { return session; }
|
||||
public void setSession(String session) { this.session = session; }
|
||||
public String getMediaMeta() { return mediaMeta; }
|
||||
public void setMediaMeta(String mediaMeta) { this.mediaMeta = mediaMeta; }
|
||||
public String getRawArchive() { return rawArchive; }
|
||||
public void setRawArchive(String rawArchive) { this.rawArchive = rawArchive; }
|
||||
public String getDlq() { return dlq; }
|
||||
public void setDlq(String dlq) { this.dlq = dlq; }
|
||||
}
|
||||
|
||||
public static class Consumer {
|
||||
/** Kafka 消费总开关。与 producer 总开关分离，默认 false，避免单体服务意外自消费。 */
|
||||
private boolean enabled = false;
|
||||
private boolean autoStartup = true;
|
||||
private String clientIdPrefix = "lingniu-envelope-consumer";
|
||||
private int pollTimeoutMillis = 1000;
|
||||
private int loopBackoffMillis = 1000;
|
||||
private String autoOffsetReset = "earliest";
|
||||
private int maxPollRecords = 500;
|
||||
private int maxPollIntervalMillis = 1800000;
|
||||
private int concurrency = 1;
|
||||
/**
|
||||
* Processor bean name -> Kafka binding。
|
||||
*
|
||||
* <p>示例 key：{@code eventHistoryEnvelopeConsumerProcessor}、
|
||||
* {@code vehicleStateEnvelopeConsumerProcessor}、{@code vehicleStatEnvelopeConsumerProcessor}。
|
||||
*/
|
||||
private Map<String, Binding> bindings = new LinkedHashMap<>();
|
||||
|
||||
public boolean isEnabled() { return enabled; }
|
||||
public void setEnabled(boolean enabled) { this.enabled = enabled; }
|
||||
public boolean isAutoStartup() { return autoStartup; }
|
||||
public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; }
|
||||
public String getClientIdPrefix() { return clientIdPrefix; }
|
||||
public void setClientIdPrefix(String clientIdPrefix) { this.clientIdPrefix = clientIdPrefix; }
|
||||
public int getPollTimeoutMillis() { return pollTimeoutMillis; }
|
||||
public void setPollTimeoutMillis(int pollTimeoutMillis) { this.pollTimeoutMillis = pollTimeoutMillis; }
|
||||
public int getLoopBackoffMillis() { return loopBackoffMillis; }
|
||||
public void setLoopBackoffMillis(int loopBackoffMillis) { this.loopBackoffMillis = loopBackoffMillis; }
|
||||
public String getAutoOffsetReset() { return autoOffsetReset; }
|
||||
public void setAutoOffsetReset(String autoOffsetReset) { this.autoOffsetReset = autoOffsetReset; }
|
||||
public int getMaxPollRecords() { return maxPollRecords; }
|
||||
public void setMaxPollRecords(int maxPollRecords) { this.maxPollRecords = maxPollRecords; }
|
||||
public int getMaxPollIntervalMillis() { return maxPollIntervalMillis; }
|
||||
public void setMaxPollIntervalMillis(int maxPollIntervalMillis) { this.maxPollIntervalMillis = maxPollIntervalMillis; }
|
||||
public int getConcurrency() { return concurrency; }
|
||||
public void setConcurrency(int concurrency) { this.concurrency = concurrency; }
|
||||
public Map<String, Binding> getBindings() { return bindings; }
|
||||
public void setBindings(Map<String, Binding> bindings) { this.bindings = bindings; }
|
||||
}
|
||||
|
||||
public static class Binding {
|
||||
/** 单个 processor binding 开关，用于临时停某个下游消费者而不关整个 consumer runner。 */
|
||||
private boolean enabled = true;
|
||||
/** Kafka consumer group id。不同服务要独立消费同一 topic 时必须使用不同 group。 */
|
||||
private String groupId;
|
||||
/** 该 processor 订阅的 topic 列表。 */
|
||||
private List<String> topics = new ArrayList<>();
|
||||
|
||||
public boolean isEnabled() { return enabled; }
|
||||
public void setEnabled(boolean enabled) { this.enabled = enabled; }
|
||||
public String getGroupId() { return groupId; }
|
||||
public void setGroupId(String groupId) { this.groupId = groupId; }
|
||||
public List<String> getTopics() { return topics; }
|
||||
public void setTopics(List<String> topics) { this.topics = topics; }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package com.lingniu.ingest.sink.kafka;
|
||||
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
|
||||
/**
|
||||
* 事件 → Kafka topic 路由。集中管理避免散落。
|
||||
*/
|
||||
public final class TopicRouter {
|
||||
|
||||
private final KafkaSinkProperties.Topics topics;
|
||||
|
||||
public TopicRouter(KafkaSinkProperties.Topics topics) {
|
||||
this.topics = topics;
|
||||
}
|
||||
|
||||
public String route(VehicleEvent event) {
|
||||
return switch (event) {
|
||||
case VehicleEvent.Realtime _ -> topics.getRealtime();
|
||||
case VehicleEvent.Location _ -> topics.getLocation();
|
||||
case VehicleEvent.Alarm _ -> topics.getAlarm();
|
||||
case VehicleEvent.Login _,
|
||||
VehicleEvent.Logout _,
|
||||
VehicleEvent.Heartbeat _ -> topics.getSession();
|
||||
case VehicleEvent.MediaMeta _ -> topics.getMediaMeta();
|
||||
case VehicleEvent.Passthrough _ -> topics.getAlarm();
|
||||
case VehicleEvent.RawArchive _ -> topics.getRawArchive();
|
||||
};
|
||||
}
|
||||
}
|
||||
168
modules/sinks/sink-kafka/src/main/proto/vehicle_envelope.proto
Normal file
168
modules/sinks/sink-kafka/src/main/proto/vehicle_envelope.proto
Normal file
@@ -0,0 +1,168 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package com.lingniu.ingest.sink.kafka.proto;
|
||||
|
||||
option java_multiple_files = true;
|
||||
option java_package = "com.lingniu.ingest.sink.kafka.proto";
|
||||
option java_outer_classname = "VehicleEnvelopeProto";
|
||||
|
||||
// 统一消息外壳：所有 Topic 共用此 Envelope，payload 通过 oneof 区分具体事件类型。
|
||||
message VehicleEnvelope {
|
||||
string schema_version = 1;
|
||||
string event_id = 2;
|
||||
string trace_id = 3;
|
||||
string vin = 4;
|
||||
string source = 5; // ProtocolId 字符串形式
|
||||
string protocol_version = 6;
|
||||
int64 event_time_ms = 7;
|
||||
int64 ingest_time_ms = 8;
|
||||
string ingest_node_id = 9;
|
||||
map<string, string> metadata = 10;
|
||||
|
||||
oneof payload {
|
||||
RealtimePayload realtime = 20;
|
||||
LocationPayload location = 21;
|
||||
AlarmPayload alarm = 22;
|
||||
LoginPayload login = 23;
|
||||
LogoutPayload logout = 24;
|
||||
HeartbeatPayload heartbeat = 25;
|
||||
MediaMetaPayload media_meta = 26;
|
||||
PassthroughPayload passthrough = 27;
|
||||
}
|
||||
|
||||
// 可选：原始字节指针
|
||||
RawArchiveRef raw_archive = 40;
|
||||
|
||||
// 全字段内部遥测快照。新下游消费者优先读取这里，避免依赖协议字段名。
|
||||
TelemetrySnapshot telemetry_snapshot = 50;
|
||||
|
||||
// 新一代事实模型：RAW 帧索引与解析事实，均不携带完整 raw bytes。
|
||||
RawFrameFactPayload raw_frame_fact = 60;
|
||||
DecodedFactPayload decoded_fact = 61;
|
||||
}
|
||||
|
||||
message RawArchiveRef {
|
||||
string uri = 1;
|
||||
string checksum = 2;
|
||||
int64 size_bytes = 3;
|
||||
string parsed_json = 4;
|
||||
}
|
||||
|
||||
message TelemetrySnapshot {
|
||||
string event_type = 1;
|
||||
string raw_archive_uri = 2;
|
||||
repeated TelemetryField fields = 3;
|
||||
}
|
||||
|
||||
message TelemetryField {
|
||||
string key = 1;
|
||||
string value_type = 2;
|
||||
string value = 3;
|
||||
string unit = 4;
|
||||
string quality = 5;
|
||||
string source_path = 6;
|
||||
}
|
||||
|
||||
enum ParseStatusProto {
|
||||
PARSE_STATUS_UNSPECIFIED = 0;
|
||||
PARSE_STATUS_NOT_PARSED = 1;
|
||||
PARSE_STATUS_SUCCEEDED = 2;
|
||||
PARSE_STATUS_FAILED = 3;
|
||||
}
|
||||
|
||||
message RawFrameFactPayload {
|
||||
string frame_id = 1;
|
||||
string vehicle_key = 2;
|
||||
string vin = 3;
|
||||
string phone = 4;
|
||||
int32 message_id = 5;
|
||||
int32 sub_type = 6;
|
||||
string raw_uri = 7;
|
||||
string checksum = 8;
|
||||
int64 raw_size_bytes = 9;
|
||||
ParseStatusProto parse_status = 10;
|
||||
string parse_error = 11;
|
||||
string peer = 12;
|
||||
map<string, string> metadata = 13;
|
||||
}
|
||||
|
||||
message DecodedFactPayload {
|
||||
string fact_id = 1;
|
||||
string frame_id = 2;
|
||||
string fact_type = 3;
|
||||
string vehicle_key = 4;
|
||||
string vin = 5;
|
||||
string phone = 6;
|
||||
string raw_uri = 7;
|
||||
map<string, string> fields = 8;
|
||||
map<string, string> metadata = 9;
|
||||
}
|
||||
|
||||
message RealtimePayload {
|
||||
optional double speed_kmh = 1;
|
||||
optional double total_mileage_km = 2;
|
||||
optional double battery_soc = 3;
|
||||
optional double battery_voltage_v = 4;
|
||||
optional double battery_current_a = 5;
|
||||
optional double fc_voltage_v = 6;
|
||||
optional double fc_current_a = 7;
|
||||
optional double fc_temp_c = 8;
|
||||
optional double hydrogen_remaining_kg = 9;
|
||||
optional double hydrogen_high_pressure_mpa = 10;
|
||||
optional double hydrogen_low_pressure_mpa = 11;
|
||||
optional string vehicle_state = 12;
|
||||
optional string charging_state = 13;
|
||||
optional string running_mode = 14;
|
||||
optional int32 gear_level = 15;
|
||||
optional double accelerator_pedal = 16;
|
||||
optional double brake_pedal = 17;
|
||||
optional double longitude = 18;
|
||||
optional double latitude = 19;
|
||||
optional double altitude_m = 20;
|
||||
optional double direction_deg = 21;
|
||||
optional double ambient_temp_c = 22;
|
||||
optional double coolant_temp_c = 23;
|
||||
}
|
||||
|
||||
message LocationPayload {
|
||||
double longitude = 1;
|
||||
double latitude = 2;
|
||||
double altitude_m = 3;
|
||||
double speed_kmh = 4;
|
||||
double direction_deg = 5;
|
||||
int64 alarm_flag = 6;
|
||||
int64 status_flag = 7;
|
||||
}
|
||||
|
||||
message AlarmPayload {
|
||||
string level = 1;
|
||||
int32 alarm_type_code = 2;
|
||||
string alarm_type_name = 3;
|
||||
repeated string fault_codes = 4;
|
||||
optional double longitude = 5;
|
||||
optional double latitude = 6;
|
||||
// 通用报警标志位（2016 版 0~15 / 2025 版 0~27），对照 GB/T 32960.3 表 24。
|
||||
// 例：SOC_LOW / BATTERY_HIGH_TEMP / HYDROGEN_LEAK
|
||||
repeated string active_bits = 7;
|
||||
}
|
||||
|
||||
message LoginPayload {
|
||||
string iccid = 1;
|
||||
string protocol_version = 2;
|
||||
}
|
||||
|
||||
message LogoutPayload {}
|
||||
|
||||
message HeartbeatPayload {}
|
||||
|
||||
message MediaMetaPayload {
|
||||
string media_id = 1;
|
||||
string media_type = 2;
|
||||
int64 size_bytes = 3;
|
||||
string archive_ref = 4;
|
||||
}
|
||||
|
||||
message PassthroughPayload {
|
||||
int32 passthrough_type = 1;
|
||||
bytes data = 2;
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
com.lingniu.ingest.sink.kafka.KafkaSinkAutoConfiguration
|
||||
com.lingniu.ingest.sink.kafka.KafkaSinkConsumerAutoConfiguration
|
||||
Reference in New Issue
Block a user