shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: initial import of lingniu-vehicle-ingest

Browse Source 
Multi-module Spring Boot ingest service for vehicle telemetry. Modules:

- ingest-api / ingest-core / ingest-codec-common: shared SPI, dispatcher,
  Disruptor event bus, BCC/BCD codec helpers
- protocol-gb32960: GB/T 32960.3 inbound (Netty + per-version parser
  packages v2016/v2025), platform login auth, VIN whitelist, idle handler
- protocol-jt808 / protocol-jt1078 / protocol-jsatl12: JT/T inbound
- inbound-mqtt / inbound-xinda-push: alternative ingest channels
- session-core: per-channel session state
- sink-archive / sink-mq: persistence sinks (local file / Kafka)
- command-gateway: terminal control command gateway
- bootstrap-all: aggregator Spring Boot app
- observability: Micrometer / Actuator wiring

Includes hex-dump golden samples under protocol-gb32960/src/test/resources
and the GB/T 32960.3-2016 / 2025 reference PDFs under reference/.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
lingniu-dev
2026-04-15 16:08:57 +08:00
commit 064ecc479c
220 changed files with 11874 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/AsyncBatchExecutor.java Normal file
View File

@@ -0,0 +1,164 @@
package com.lingniu.ingest.core.concurrency;
import com.lingniu.ingest.api.annotation.AsyncBatch;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.core.dispatcher.HandlerDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
/**
* {@link AsyncBatch} 注解的执行器：把 Handler 方法从"单条调用"变成"批量调用"。
*
* <p>每个 Handler 方法对应一个 {@link Batcher}，背后是一个固定容量的 {@link BlockingQueue}
* 和 {@link AsyncBatch#poolSize()} 条守护虚拟线程。批量触发条件：
* <ul>
* <li>累积 {@link AsyncBatch#size()} 条立即 flush
* <li>从第一条入队起等待 {@link AsyncBatch#waitMs()} 毫秒后强制 flush
* </ul>
*
* <p>目标 Handler 方法签名约定：
* <pre>{@code
* @MessageMapping(...)
* @AsyncBatch(size = 4000, waitMs = 1000, poolSize = 2)
* public List<VehicleEvent> onBatch(List<PayloadType> batch) { ... }
* }</pre>
*
* <p>flush 产出的事件由构造器注入的 {@code eventPublisher} 异步投递（通常是
* {@link DisruptorEventBus#publish}），不阻塞 Netty EventLoop。
*/
public final class AsyncBatchExecutor implements AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(AsyncBatchExecutor.class);
private final Consumer<VehicleEvent> eventPublisher;
private final ConcurrentMap<Method, Batcher> batchers = new ConcurrentHashMap<>();
public AsyncBatchExecutor(Consumer<VehicleEvent> eventPublisher) {
this.eventPublisher = eventPublisher;
}
/** 供 Dispatcher 调用：把单条消息交给目标 Handler 的 batcher。 */
public void submit(HandlerDefinition def, Object message) {
Batcher b = batchers.computeIfAbsent(def.method(), m -> new Batcher(def, eventPublisher));
b.offer(message);
}
@Override
public void close() {
batchers.values().forEach(Batcher::close);
batchers.clear();
log.info("AsyncBatchExecutor closed");
}
// ===== internals =====
private static final class Batcher implements AutoCloseable {
private final HandlerDefinition def;
private final Consumer<VehicleEvent> publisher;
private final int batchSize;
private final long waitMs;
private final BlockingQueue<Object> queue;
private final Thread[] workers;
private volatile boolean running = true;
Batcher(HandlerDefinition def, Consumer<VehicleEvent> publisher) {
AsyncBatch cfg = def.asyncBatch();
this.def = def;
this.publisher = publisher;
this.batchSize = Math.max(1, cfg.size());
this.waitMs = Math.max(1, cfg.waitMs());
this.queue = new ArrayBlockingQueue<>(Math.max(batchSize * 4, 16));
int pool = Math.max(1, cfg.poolSize());
this.workers = new Thread[pool];
for (int i = 0; i < pool; i++) {
Thread t = Thread.ofVirtual()
.name("batcher-" + def.method().getName() + "-" + i)
.unstarted(this::loop);
workers[i] = t;
t.start();
}
log.info("batcher started method={} size={} waitMs={} pool={}",
def.method().getName(), batchSize, waitMs, pool);
}
void offer(Object item) {
try {
if (!queue.offer(item, 100, TimeUnit.MILLISECONDS)) {
log.warn("batcher queue full, dropping item for {}", def.method().getName());
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void loop() {
while (running) {
try {
Object first = queue.poll(200, TimeUnit.MILLISECONDS);
if (first == null) continue;
List<Object> buf = new ArrayList<>(batchSize);
buf.add(first);
long deadlineNanos = System.nanoTime() + waitMs * 1_000_000L;
while (buf.size() < batchSize) {
long remain = deadlineNanos - System.nanoTime();
if (remain <= 0) break;
Object next = queue.poll(remain, TimeUnit.NANOSECONDS);
if (next == null) break;
buf.add(next);
}
flush(buf);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
} catch (Exception e) {
log.error("batcher loop error method={}", def.method().getName(), e);
}
}
}
@SuppressWarnings("unchecked")
private void flush(List<Object> buf) {
List<VehicleEvent> events;
try {
Object result = def.method().invoke(def.bean(), buf);
events = switch (result) {
case null -> List.of();
case List<?> list -> (List<VehicleEvent>) list;
case VehicleEvent e -> List.of(e);
default -> throw new IllegalStateException(
"@AsyncBatch method must return List<VehicleEvent>: " + def.method());
};
} catch (InvocationTargetException e) {
log.error("batcher invoke failed method={}", def.method().getName(), e.getCause());
return;
} catch (IllegalAccessException e) {
log.error("batcher access denied method={}", def.method().getName(), e);
return;
}
for (VehicleEvent e : events) {
try {
publisher.accept(e);
} catch (Exception ex) {
log.warn("batcher publish failed eventId={}", e.eventId(), ex);
}
}
}
@Override
public void close() {
running = false;
for (Thread t : workers) t.interrupt();
}
}
}

87
ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java Normal file
View File

@@ -0,0 +1,87 @@
package com.lingniu.ingest.core.concurrency;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.sink.EventSink;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;
/**
* 基于 Disruptor 的事件总线：Dispatcher 投递事件 → RingBuffer → Sink 扇出。
*
* <p>使用虚拟线程作为 Sink 消费线程，避免阻塞 IO 占用平台线程。
* 单 VIN 有序性由上游 Netty + 分区投递保证，RingBuffer 不做 per-vin 排序。
*/
public final class DisruptorEventBus implements AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(DisruptorEventBus.class);
private final Disruptor<VehicleEventSlot> disruptor;
private final AtomicLong published = new AtomicLong();
private final AtomicLong failed = new AtomicLong();
public DisruptorEventBus(int ringBufferSize, String waitStrategyName, List<EventSink> sinks) {
ThreadFactory tf = Thread.ofVirtual().name("ingest-bus-", 0).factory();
this.disruptor = new Disruptor<>(
VehicleEventSlot::new,
ringBufferSize,
tf,
ProducerType.MULTI,
waitStrategy(waitStrategyName));
EventHandler<VehicleEventSlot>[] handlers = sinks.stream()
.map(this::toHandler)
.toArray(EventHandler[]::new);
disruptor.handleEventsWith(handlers);
disruptor.start();
log.info("DisruptorEventBus started ringBuffer={} wait={} sinks={}",
ringBufferSize, waitStrategyName, sinks.size());
}
public void publish(VehicleEvent event) {
disruptor.getRingBuffer().publishEvent((slot, seq, e) -> slot.event = e, event);
published.incrementAndGet();
}
@Override
public void close() {
disruptor.shutdown();
log.info("DisruptorEventBus stopped published={} failed={}", published.get(), failed.get());
}
private EventHandler<VehicleEventSlot> toHandler(EventSink sink) {
return (slot, seq, endOfBatch) -> {
VehicleEvent e = slot.event;
if (e == null || !sink.accepts(e)) return;
try {
sink.publish(e).exceptionally(ex -> {
failed.incrementAndGet();
log.warn("sink {} publish failed", sink.name(), ex);
return null;
});
} finally {
if (endOfBatch) slot.clear();
}
};
}
private static WaitStrategy waitStrategy(String name) {
return switch (name == null ? "yielding" : name.toLowerCase()) {
case "blocking" -> new BlockingWaitStrategy();
case "sleeping" -> new SleepingWaitStrategy();
case "busy-spin" -> new BusySpinWaitStrategy();
default -> new YieldingWaitStrategy();
};
}
}

12
ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/VehicleEventSlot.java Normal file
View File

@@ -0,0 +1,12 @@
package com.lingniu.ingest.core.concurrency;
import com.lingniu.ingest.api.event.VehicleEvent;
/** Disruptor RingBuffer 槽位：持有可变引用，避免每次发布都分配新对象。 */
public final class VehicleEventSlot {
public VehicleEvent event;
public void clear() {
this.event = null;
}
}

87
ingest-core/src/main/java/com/lingniu/ingest/core/config/IngestCoreAutoConfiguration.java Normal file
View File

@@ -0,0 +1,87 @@
package com.lingniu.ingest.core.config;
import com.lingniu.ingest.api.pipeline.IngestInterceptor;
import com.lingniu.ingest.api.sink.EventSink;
import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor;
import com.lingniu.ingest.core.concurrency.DisruptorEventBus;
import com.lingniu.ingest.core.dispatcher.AnnotationHandlerBeanPostProcessor;
import com.lingniu.ingest.core.dispatcher.Dispatcher;
import com.lingniu.ingest.core.dispatcher.HandlerInvoker;
import com.lingniu.ingest.core.dispatcher.HandlerRegistry;
import com.lingniu.ingest.core.pipeline.InterceptorChain;
import com.lingniu.ingest.core.pipeline.builtin.DedupInterceptor;
import com.lingniu.ingest.core.pipeline.builtin.RateLimitInterceptor;
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.util.List;
/**
* ingest-core 的自动装配入口。所有 Bean 都是 {@code @ConditionalOnMissingBean}，便于下游替换。
*/
@AutoConfiguration
@EnableConfigurationProperties(IngestCoreProperties.class)
public class IngestCoreAutoConfiguration {
@Bean
@ConditionalOnMissingBean
public HandlerRegistry handlerRegistry() {
return new HandlerRegistry();
}
@Bean
@ConditionalOnMissingBean
public HandlerInvoker handlerInvoker() {
return new HandlerInvoker();
}
@Bean
public AnnotationHandlerBeanPostProcessor annotationHandlerBeanPostProcessor(HandlerRegistry registry) {
return new AnnotationHandlerBeanPostProcessor(registry);
}
@Bean
@ConditionalOnProperty(prefix = "lingniu.ingest.pipeline.dedup", name = "enabled", havingValue = "true", matchIfMissing = true)
public DedupInterceptor dedupInterceptor(IngestCoreProperties props) {
return new DedupInterceptor(props.getDedup().getCacheSize(), props.getDedup().getTtlSeconds());
}
@Bean
public RateLimitInterceptor rateLimitInterceptor(IngestCoreProperties props) {
return new RateLimitInterceptor(props.getRateLimit().getPerVinQps(), props.getRateLimit().getMaxVins());
}
@Bean
@ConditionalOnMissingBean
public InterceptorChain interceptorChain(List<IngestInterceptor> interceptors) {
return new InterceptorChain(interceptors);
}
@Bean(destroyMethod = "close")
@ConditionalOnMissingBean
public DisruptorEventBus disruptorEventBus(IngestCoreProperties props, List<EventSink> sinks) {
return new DisruptorEventBus(
props.getDisruptor().getRingBufferSize(),
props.getDisruptor().getWaitStrategy(),
sinks);
}
@Bean(destroyMethod = "close")
@ConditionalOnMissingBean
public AsyncBatchExecutor asyncBatchExecutor(DisruptorEventBus eventBus) {
return new AsyncBatchExecutor(eventBus::publish);
}
@Bean
@ConditionalOnMissingBean
public Dispatcher dispatcher(HandlerRegistry registry,
InterceptorChain chain,
HandlerInvoker invoker,
DisruptorEventBus eventBus,
AsyncBatchExecutor batchExecutor) {
return new Dispatcher(registry, chain, invoker, eventBus, batchExecutor);
}
}

56
ingest-core/src/main/java/com/lingniu/ingest/core/config/IngestCoreProperties.java Normal file
View File

@@ -0,0 +1,56 @@
package com.lingniu.ingest.core.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
@ConfigurationProperties(prefix = "lingniu.ingest.pipeline")
public class IngestCoreProperties {
private Disruptor disruptor = new Disruptor();
private Dedup dedup = new Dedup();
private RateLimit rateLimit = new RateLimit();
public Disruptor getDisruptor() { return disruptor; }
public void setDisruptor(Disruptor disruptor) { this.disruptor = disruptor; }
public Dedup getDedup() { return dedup; }
public void setDedup(Dedup dedup) { this.dedup = dedup; }
public RateLimit getRateLimit() { return rateLimit; }
public void setRateLimit(RateLimit rateLimit) { this.rateLimit = rateLimit; }
public static class Disruptor {
private int ringBufferSize = 131072;
private String waitStrategy = "yielding";
private String producerType = "multi";
public int getRingBufferSize() { return ringBufferSize; }
public void setRingBufferSize(int ringBufferSize) { this.ringBufferSize = ringBufferSize; }
public String getWaitStrategy() { return waitStrategy; }
public void setWaitStrategy(String waitStrategy) { this.waitStrategy = waitStrategy; }
public String getProducerType() { return producerType; }
public void setProducerType(String producerType) { this.producerType = producerType; }
}
public static class Dedup {
private boolean enabled = true;
private int cacheSize = 200000;
private long ttlSeconds = 600;
public boolean isEnabled() { return enabled; }
public void setEnabled(boolean enabled) { this.enabled = enabled; }
public int getCacheSize() { return cacheSize; }
public void setCacheSize(int cacheSize) { this.cacheSize = cacheSize; }
public long getTtlSeconds() { return ttlSeconds; }
public void setTtlSeconds(long ttlSeconds) { this.ttlSeconds = ttlSeconds; }
}
public static class RateLimit {
private int perVinQps = 50;
private int maxVins = 100000;
public int getPerVinQps() { return perVinQps; }
public void setPerVinQps(int perVinQps) { this.perVinQps = perVinQps; }
public int getMaxVins() { return maxVins; }
public void setMaxVins(int maxVins) { this.maxVins = maxVins; }
}
}

64
ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/AnnotationHandlerBeanPostProcessor.java Normal file
View File

@@ -0,0 +1,64 @@
package com.lingniu.ingest.core.dispatcher;
import com.lingniu.ingest.api.annotation.AsyncBatch;
import com.lingniu.ingest.api.annotation.IdempotentKey;
import com.lingniu.ingest.api.annotation.MessageMapping;
import com.lingniu.ingest.api.annotation.ProtocolHandler;
import com.lingniu.ingest.api.annotation.RateLimited;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.annotation.AnnotatedElementUtils;
import java.lang.reflect.Method;
/**
* Spring BeanPostProcessor：扫描带 {@link ProtocolHandler} 注解的 Bean，
* 把每个带 {@link MessageMapping} 的方法注册到 {@link HandlerRegistry}。
*
* <p>替代旧代码里遍布的 {@code if (msgId == 0x0100) ... else if (msgId == 0x0102) ...} 风格。
*/
public class AnnotationHandlerBeanPostProcessor implements BeanPostProcessor {
private static final Logger log = LoggerFactory.getLogger(AnnotationHandlerBeanPostProcessor.class);
private final HandlerRegistry registry;
public AnnotationHandlerBeanPostProcessor(HandlerRegistry registry) {
this.registry = registry;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
Class<?> type = bean.getClass();
ProtocolHandler classAnno = AnnotatedElementUtils.findMergedAnnotation(type, ProtocolHandler.class);
if (classAnno == null) return bean;
for (Method method : type.getMethods()) {
MessageMapping mapping = AnnotatedElementUtils.findMergedAnnotation(method, MessageMapping.class);
if (mapping == null) continue;
Class<?> paramType = method.getParameterCount() > 0 ? method.getParameterTypes()[0] : Object.class;
int[] commands = mapping.command().length == 0 ? new int[]{0} : mapping.command();
int[] infoTypes = mapping.infoType().length == 0 ? new int[]{0} : mapping.infoType();
RateLimited rl = AnnotatedElementUtils.findMergedAnnotation(method, RateLimited.class);
IdempotentKey ik = AnnotatedElementUtils.findMergedAnnotation(method, IdempotentKey.class);
AsyncBatch ab = AnnotatedElementUtils.findMergedAnnotation(method, AsyncBatch.class);
for (int cmd : commands) {
for (int info : infoTypes) {
HandlerDefinition def = new HandlerDefinition(
classAnno.protocol(), cmd, info, mapping.desc(),
bean, method, paramType, rl, ik, ab);
registry.register(def);
log.info("registered handler {} protocol={} command=0x{} info=0x{}",
type.getSimpleName() + "#" + method.getName(),
classAnno.protocol(), Integer.toHexString(cmd), Integer.toHexString(info));
}
}
}
return bean;
}
}

79
ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java Normal file
View File

@@ -0,0 +1,79 @@
package com.lingniu.ingest.core.dispatcher;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.pipeline.IngestContext;
import com.lingniu.ingest.api.pipeline.RawFrame;
import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor;
import com.lingniu.ingest.core.concurrency.DisruptorEventBus;
import com.lingniu.ingest.core.pipeline.InterceptorChain;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.UUID;
/**
* 核心分发器：RawFrame → 拦截链 → Handler → 事件 → Disruptor EventBus。
*
* <p>所有 Inbound Adapter（Netty / MQTT / PushClient）都调用 {@link #dispatch(RawFrame)}，
* 协议差异在这里被彻底抹平。
*
* <p>对于带 {@code @AsyncBatch} 的 Handler，Dispatcher 不直接反射调用，而是把消息交给
* {@link AsyncBatchExecutor} 进行聚合 → 批量调用 → 异步发布事件。
*/
public final class Dispatcher {
private static final Logger log = LoggerFactory.getLogger(Dispatcher.class);
private final HandlerRegistry registry;
private final InterceptorChain interceptors;
private final HandlerInvoker invoker;
private final DisruptorEventBus eventBus;
private final AsyncBatchExecutor batchExecutor;
public Dispatcher(HandlerRegistry registry,
InterceptorChain interceptors,
HandlerInvoker invoker,
DisruptorEventBus eventBus,
AsyncBatchExecutor batchExecutor) {
this.registry = registry;
this.interceptors = interceptors;
this.invoker = invoker;
this.eventBus = eventBus;
this.batchExecutor = batchExecutor;
}
public void dispatch(RawFrame frame) {
IngestContext ctx = new IngestContext(UUID.randomUUID().toString());
try {
if (!interceptors.before(frame, ctx)) {
log.debug("frame aborted: {}", ctx.abortReason());
return;
}
List<HandlerDefinition> handlers = registry.resolve(
frame.protocolId(), frame.command(), frame.infoType());
if (handlers.isEmpty()) {
log.debug("no handler for {} cmd=0x{} info=0x{}",
frame.protocolId(), Integer.toHexString(frame.command()),
Integer.toHexString(frame.infoType()));
return;
}
for (HandlerDefinition def : handlers) {
if (def.asyncBatch() != null) {
batchExecutor.submit(def, frame.payload());
continue;
}
List<VehicleEvent> events = invoker.invoke(def, frame, ctx);
for (VehicleEvent e : events) {
interceptors.after(e, ctx);
eventBus.publish(e);
}
}
} catch (Throwable t) {
log.error("dispatch failure traceId={}", ctx.traceId(), t);
interceptors.onError(t, ctx);
}
}
}

31
ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/HandlerDefinition.java Normal file
View File

@@ -0,0 +1,31 @@
package com.lingniu.ingest.core.dispatcher;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.annotation.AsyncBatch;
import com.lingniu.ingest.api.annotation.IdempotentKey;
import com.lingniu.ingest.api.annotation.RateLimited;
import java.lang.reflect.Method;
/**
* 一个被注解扫描到的 Handler 方法的静态描述。不可变。
*/
public record HandlerDefinition(
ProtocolId protocol,
int command,
int infoType,
String desc,
Object bean,
Method method,
Class<?> parameterType,
RateLimited rateLimited,
IdempotentKey idempotentKey,
AsyncBatch asyncBatch
) {
public boolean matches(ProtocolId p, int cmd, int info) {
if (p != protocol) return false;
if (command != 0 && command != cmd) return false;
if (infoType != 0 && infoType != info) return false;
return true;
}
}

33
ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/HandlerInvoker.java Normal file
View File

@@ -0,0 +1,33 @@
package com.lingniu.ingest.core.dispatcher;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.pipeline.IngestContext;
import com.lingniu.ingest.api.pipeline.RawFrame;
import java.lang.reflect.InvocationTargetException;
import java.util.Collections;
import java.util.List;
/**
* 反射调用 Handler。单独抽出是为了后续可插拔：未来可替换为 MethodHandle 或字节码生成。
*/
public class HandlerInvoker {
@SuppressWarnings("unchecked")
public List<VehicleEvent> invoke(HandlerDefinition def, RawFrame frame, IngestContext ctx) {
try {
Object result = def.method().invoke(def.bean(), frame.payload());
return switch (result) {
case null -> Collections.emptyList();
case VehicleEvent e -> List.of(e);
case List<?> list -> (List<VehicleEvent>) list;
default -> throw new IllegalStateException(
"Handler return type must be VehicleEvent or List<VehicleEvent>: " + def.method());
};
} catch (InvocationTargetException e) {
throw new RuntimeException("Handler threw exception: " + def.method(), e.getCause());
} catch (IllegalAccessException e) {
throw new RuntimeException("Handler not accessible: " + def.method(), e);
}
}
}

48
ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/HandlerRegistry.java Normal file
View File

@@ -0,0 +1,48 @@
package com.lingniu.ingest.core.dispatcher;
import com.lingniu.ingest.api.ProtocolId;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Handler 注册中心。按 {@code (ProtocolId, command)} 索引，O(1) 查找；同 key 可注册多个 Handler，由
* {@link HandlerDefinition#matches} 进一步精确匹配 {@code infoType}。
*/
public final class HandlerRegistry {
private final Map<RoutingKey, List<HandlerDefinition>> byRoute = new ConcurrentHashMap<>();
private final List<HandlerDefinition> wildcardHandlers = Collections.synchronizedList(new ArrayList<>());
public void register(HandlerDefinition def) {
if (def.command() == 0) {
wildcardHandlers.add(def);
return;
}
byRoute.computeIfAbsent(new RoutingKey(def.protocol(), def.command()),
k -> Collections.synchronizedList(new ArrayList<>())).add(def);
}
public List<HandlerDefinition> resolve(ProtocolId protocol, int command, int infoType) {
List<HandlerDefinition> exact = byRoute.get(new RoutingKey(protocol, command));
List<HandlerDefinition> result = new ArrayList<>();
if (exact != null) {
for (HandlerDefinition d : exact) {
if (d.matches(protocol, command, infoType)) result.add(d);
}
}
for (HandlerDefinition d : wildcardHandlers) {
if (d.matches(protocol, command, infoType)) result.add(d);
}
return result;
}
public int size() {
return byRoute.values().stream().mapToInt(List::size).sum() + wildcardHandlers.size();
}
private record RoutingKey(ProtocolId protocol, int command) {}
}

45
ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/InterceptorChain.java Normal file
View File

@@ -0,0 +1,45 @@
package com.lingniu.ingest.core.pipeline;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.pipeline.IngestContext;
import com.lingniu.ingest.api.pipeline.IngestInterceptor;
import com.lingniu.ingest.api.pipeline.RawFrame;
import org.springframework.core.annotation.AnnotationAwareOrderComparator;
import java.util.ArrayList;
import java.util.List;
/**
* 顺序执行的拦截器链，通过 Spring {@code @Order} 或 {@code Ordered} 排序。
*/
public final class InterceptorChain {
private final List<IngestInterceptor> interceptors;
public InterceptorChain(List<IngestInterceptor> interceptors) {
List<IngestInterceptor> sorted = new ArrayList<>(interceptors);
AnnotationAwareOrderComparator.sort(sorted);
this.interceptors = List.copyOf(sorted);
}
public boolean before(RawFrame frame, IngestContext ctx) {
for (IngestInterceptor i : interceptors) {
if (!i.before(frame, ctx) || ctx.aborted()) {
return false;
}
}
return true;
}
public void after(VehicleEvent event, IngestContext ctx) {
for (IngestInterceptor i : interceptors) {
i.after(event, ctx);
}
}
public void onError(Throwable error, IngestContext ctx) {
for (IngestInterceptor i : interceptors) {
i.onError(error, ctx);
}
}
}

45
ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/DedupInterceptor.java Normal file
View File

@@ -0,0 +1,45 @@
package com.lingniu.ingest.core.pipeline.builtin;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.lingniu.ingest.api.pipeline.IngestContext;
import com.lingniu.ingest.api.pipeline.IngestInterceptor;
import com.lingniu.ingest.api.pipeline.RawFrame;
import org.springframework.core.Ordered;
import java.time.Duration;
/**
* 基于 Caffeine 的本地幂等去重。
*
* <p>Key 构造：{@code protocolId + command + sourceMeta.seq} —— 真实实现可以接入 Redis 做多节点一致性，
* 本实现只覆盖单节点场景，满足第一阶段 PoC 需求。
*/
public class DedupInterceptor implements IngestInterceptor, Ordered {
private final Cache<String, Boolean> seen;
public DedupInterceptor(int maxSize, long ttlSeconds) {
this.seen = Caffeine.newBuilder()
.maximumSize(maxSize)
.expireAfterWrite(Duration.ofSeconds(ttlSeconds))
.build();
}
@Override
public boolean before(RawFrame frame, IngestContext ctx) {
String vin = frame.sourceMeta().getOrDefault("vin", "unknown");
String seq = frame.sourceMeta().getOrDefault("seq", "0");
String key = frame.protocolId() + ":" + vin + ":" + frame.command() + ":" + seq;
if (seen.asMap().putIfAbsent(key, Boolean.TRUE) != null) {
ctx.abort("duplicate:" + key);
return false;
}
return true;
}
@Override
public int getOrder() {
return 100;
}
}

49
ingest-core/src/main/java/com/lingniu/ingest/core/pipeline/builtin/RateLimitInterceptor.java Normal file
View File

@@ -0,0 +1,49 @@
package com.lingniu.ingest.core.pipeline.builtin;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.lingniu.ingest.api.pipeline.IngestContext;
import com.lingniu.ingest.api.pipeline.IngestInterceptor;
import com.lingniu.ingest.api.pipeline.RawFrame;
import io.github.resilience4j.ratelimiter.RateLimiter;
import io.github.resilience4j.ratelimiter.RateLimiterConfig;
import org.springframework.core.Ordered;
import java.time.Duration;
/**
* 单 VIN 速率限制。每个 VIN 一个独立的 Resilience4j RateLimiter，由 Caffeine 按 LRU 管理。
*/
public class RateLimitInterceptor implements IngestInterceptor, Ordered {
private final Cache<String, RateLimiter> limiters;
private final RateLimiterConfig defaultConfig;
public RateLimitInterceptor(int perVinQps, int maxVins) {
this.defaultConfig = RateLimiterConfig.custom()
.limitForPeriod(perVinQps)
.limitRefreshPeriod(Duration.ofSeconds(1))
.timeoutDuration(Duration.ZERO)
.build();
this.limiters = Caffeine.newBuilder()
.maximumSize(maxVins)
.expireAfterAccess(Duration.ofMinutes(10))
.build();
}
@Override
public boolean before(RawFrame frame, IngestContext ctx) {
String vin = frame.sourceMeta().getOrDefault("vin", "unknown");
RateLimiter rl = limiters.get(vin, k -> RateLimiter.of("vin-" + k, defaultConfig));
if (!rl.acquirePermission()) {
ctx.abort("rate-limited:" + vin);
return false;
}
return true;
}
@Override
public int getOrder() {
return 200;
}
}

1
ingest-core/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
com.lingniu.ingest.core.config.IngestCoreAutoConfiguration