4. 状态存储方案选型：内存Map、Redis、消息队列Kafka、关系型数据库对比

在模块边界通信中，埋点状态通常需要临时存储以便后续聚合、重试或审计。选择合适的存储方案直接影响系统的吞吐、延迟、可靠性与运维成本。本节从读写性能、数据持久化、分布式能力、状态生命周期四个核心维度对比四种典型方案，并给出适用场景与实现要点。

4.1 核心维度对比

4.2 内存 Map（单机瞬态存储）

原理：在进程内维护 map[string]interface{} 或 ConcurrentHashMap，埋点状态以 key-value 形式驻留在堆内存中。适用场景：单机并行模块间传递状态，生命周期短（毫秒级），不要求跨进程共享。

实现流程（Go 语言示例）：

type InMemoryStore struct {
    mu   sync.RWMutex
    data map[string]*StateContainer
}

func NewInMemoryStore() *InMemoryStore {
    return &InMemoryStore{data: make(map[string]*StateContainer)}
}

func (s *InMemoryStore) Set(key string, v *StateContainer) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.data[key] = v
}

func (s *InMemoryStore) Get(key string) (*StateContainer, bool) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    v, ok := s.data[key]
    return v, ok
}

func (s *InMemoryStore) Delete(key string) {
    s.mu.Lock()
    defer s.mu.Unlock()
    delete(s.data, key)
}

常见错误：多协程并发读写未加锁导致 panic；GC 压力大时频繁扩容 map 造成停顿。

4.3 Redis（分布式内存状态）

原理：通过远端 Redis 实例存储 JSON 序列化后的埋点状态，利用 TTL 自动过期。适合跨进程、跨机器共享，例如多个工作节点都需访问同一批次状态。

配置示例（连接池 + 序列化）：

import redis
import pickle

pool = redis.ConnectionPool(host='127.0.0.1', port=6379, db=0, max_connections=50)
r = redis.Redis(connection_pool=pool)

# 写入状态，TTL 30秒
r.setex("batch:status:abc123", 30, pickle.dumps(state_obj))

# 读取
data = r.get("batch:status:abc123")
if data:
    state = pickle.loads(data)

调试命令：redis-cli TTL batch:status:abc123 查看剩余过期时间。

稳定性优化：开启 AOF 持久化并设置 appendfsync everysec；使用哨兵或集群保障 HA。

4.4 Kafka（流式状态与回放）

原理：将埋点状态视为事件流写入 Kafka topic，消费者按偏移量消费状态，适合需要回溯、异步解耦的场景。Kafka 同时承担存储与传输，状态可保留到指定时间或直到用户消费确认。

生产者示例（Java）：

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
props.put("acks", "all");

KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);

StateContainer state = new StateContainer(batchId, currentSeq, payload);
ProducerRecord<String, byte[]> record = new ProducerRecord<>("state-events", batchId, serialize(state));
producer.send(record).get();

消费者配置：

props.put("group.id", "state-processor");
props.put("enable.auto.commit", "false");  // 手动提交保证至少一次
props.put("auto.offset.reset", "earliest");
props.put("max.poll.records", "100");

while (true) {
    ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(200));
    for (ConsumerRecord<String, byte[]> record : records) {
        StateContainer state = deserialize(record.value());
        process(state);
    }
    consumer.commitSync();
}

常见错误：未处理 rebalance 导致瞬态重复消费；消息体过大引起分区 OOM。

4.5 关系型数据库（持久化与复杂查询）

原理：将埋点状态写入关系库表（如 MySQL/PostgreSQL），通过唯一索引保证批次内状态不冲突。适合需要事务更新、事后统计分析或审计的强持久化场景。

表结构 DDL：

CREATE TABLE t_embedded_state (
    id           BIGINT AUTO_INCREMENT PRIMARY KEY,
    batch_id     VARCHAR(64) NOT NULL,
    module       VARCHAR(32) NOT NULL,
    seq_no       INT NOT NULL,
    state_json   JSON NOT NULL,
    created_at   TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_batch_seq (batch_id, seq_no)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

写入流程（Python + psycopg2）：

import psycopg2

conn = psycopg2.connect("dbname=monitor user=writer")
cur = conn.cursor()
cur.execute("""
    INSERT INTO t_embedded_state (batch_id, module, seq_no, state_json)
    VALUES (%s, %s, %s, %s)
    ON CONFLICT (batch_id, seq_no) DO NOTHING
""", (batch_id, "ingester", seq, json.dumps(state)))
conn.commit()

性能瓶颈：频繁写事务导致死锁或锁等待；state_json 字段查询效率低。优化方向：批量插入（每批 100~500 行），使用 COPY 协议。

4.6 选型决策流程

下图给出了根据是否需要持久化、是否跨进程、是否需要回放三个条件进行选型的决策路径。