从零构建高效Skill系统：架构设计与性能优化实战

3次阅读

共计 2984 个字符，预计需要花费 8 分钟才能阅读完成。

在构建 Skill 系统时，我们常常会遇到一些性能问题。传统的 Skill 系统大多采用同步轮询模式，这种设计在并发请求量不大的情况下还能应付，但随着用户数量的增长，问题就逐渐暴露出来了。

线程阻塞问题 ：每个请求都需要独立的线程处理，当并发量高时，线程池很快就会被耗尽
状态同步困难 ：多个服务实例之间的状态难以保持一致，容易产生脏数据
响应延迟高 ：同步调用的链路过长，导致整体响应时间不可控
扩展性差 ：垂直扩展成本高，水平扩展又面临状态同步的难题

为了解决上述问题，我们引入了事件驱动架构。下面是一组基准测试数据对比：

架构类型	100 并发 QPS	1000 并发 QPS	资源占用
轮询模式	1,200	系统崩溃	高
事件驱动	8,500	15,000	中

从数据可以看出，事件驱动架构在高并发场景下优势明显。

// 生产者配置
@Configuration
public class KafkaProducerConfig {@Value("${kafka.bootstrap.servers}")
    private String bootstrapServers;

    @Bean
    public ProducerFactory<String, SkillEvent> producerFactory() {Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        // 开启重试机制
        configProps.put(ProducerConfig.RETRIES_CONFIG, 3);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, SkillEvent> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());
    }
}

// 事件序列化（使用 JSON）public class SkillEvent {
    private String eventId;
    private String skillId;
    private EventType type;
    private Map<String, Object> payload;
    private Instant timestamp;
    // 省略 getter/setter
}

@KafkaListener(topics = "skill-events", groupId = "skill-processor")
public void consume(SkillEvent event, Acknowledgment ack) {
    try {
        // 幂等检查
        if (eventLogRepository.existsByEventId(event.getEventId())) {log.warn("Duplicate event detected: {}", event.getEventId());
            ack.acknowledge();
            return;
        }

        // 业务处理
        skillService.processEvent(event);

        // 记录已处理事件
        eventLogRepository.save(new EventLog(event));
        ack.acknowledge();} catch (Exception e) {log.error("Failed to process event: {}", event.getEventId(), e);
        // 不确认消息，触发重试
    }
}

// 死信队列配置
@Bean
public DeadLetterPublishingRecoverer dlqRecoverer(KafkaOperations<String, Object> template) {
    return new DeadLetterPublishingRecoverer(template, 
        (record, ex) -> new TopicPartition("skill-events.DLQ", 0));
}

合理的事件分区可以显著提升并行处理能力：

基于 skillId 分区 ：确保同一 skill 的事件始终由同一消费者处理
热点分散 ：对高频 skill 采用 hash 取模分散到不同分区
动态调整 ：根据监控数据实时调整分区数量

// 使用 Caffeine 实现本地缓存
@Bean
public Cache<String, SkillState> skillStateCache() {return Caffeine.newBuilder()
        .maximumSize(10_000)
        .expireAfterWrite(5, TimeUnit.MINUTES)
        .build();}

// 缓存更新策略
public void updateCache(SkillEvent event) {String cacheKey = "skill:" + event.getSkillId();
    skillStateCache.asMap().compute(cacheKey, (k, v) -> {if (v == null) return loadFromDb(event.getSkillId());
        return applyEvent(v, event);
    });

    // 异步同步到数据库
    asyncExecutor.execute(() -> {SkillState state = skillStateCache.getIfPresent(cacheKey);
        if (state != null) {skillRepository.save(state);
        }
    });
}