第 205 章:中间件综合实战

学习目标

• 综合使用 Redis / MQ / ES / MongoDB
• 实现订单完整流程
• 学会多数据源架构
• 理解 CAP 理论

一、综合中间件架构

二、CAP 理论

系统	CAP	例子
MySQL	CA	单机
ZooKeeper	CP	强一致
Eureka	AP	高可用
Redis	AP	缓存
MongoDB	CP	默认

三、订单完整流程

3.1 架构

3.2 核心代码

@Service
public class OrderService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Autowired
    private RedissonClient redisson;

    @Transactional
    public Order createOrder(Long userId, OrderRequest req) {
        // 1. 防重(幂等)
        String idempotencyKey = "order:idempotent:" + req.getIdempotencyId();
        if (!redisTemplate.opsForValue().setIfAbsent(idempotencyKey, "1", 24, TimeUnit.HOURS)) {
            throw new BusinessException("请勿重复提交");
        }

        // 2. 分布式锁(防超卖)
        RLock lock = redisson.getLock("seckill:" + req.getProductId());
        try {
            if (!lock.tryLock(3, 10, TimeUnit.SECONDS)) {
                throw new BusinessException("系统繁忙");
            }

            // 3. 减库存(Lua 原子)
            Long stock = redisTemplate.execute(stockScript,
                Collections.singletonList("stock:" + req.getProductId()));
            if (stock == null || stock < req.getQuantity()) {
                throw new BusinessException("库存不足");
            }

            // 4. 写 DB
            Order order = new Order();
            order.setUserId(userId);
            order.setProductId(req.getProductId());
            order.setQuantity(req.getQuantity());
            order.setAmount(req.getAmount());
            order.setStatus("CREATED");
            orderMapper.insert(order);

            // 5. 发 MQ
            kafkaTemplate.send("order-events", order.getId().toString(),
                OrderEvent.from(order));

            return order;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("系统异常");
        } finally {
            lock.unlock();
        }
    }
}

四、多级缓存

4.1 实现

@Service
public class UserCacheService {

    // L1: 本地缓存
    private final LoadingCache<Long, User> localCache = Caffeine.newBuilder()
        .maximumSize(10_000)
        .expireAfterWrite(5, TimeUnit.MINUTES)
        .build(this::loadFromRedis);

    @Autowired
    private RedisTemplate<String, User> redisTemplate;

    @Autowired
    private UserMapper userMapper;

    public User findById(Long id) {
        return localCache.get(id);
    }

    private User loadFromRedis(Long id) {
        // L2: Redis
        String key = "user:" + id;
        User user = redisTemplate.opsForValue().get(key);
        if (user != null) return user;

        // L3: DB
        user = userMapper.selectById(id);
        if (user != null) {
            redisTemplate.opsForValue().set(key, user, 30, TimeUnit.MINUTES);
        }
        return user;
    }

    public void invalidate(Long id) {
        localCache.invalidate(id);
        redisTemplate.delete("user:" + id);
    }
}

五、数据同步

5.1 DB → ES(Canal)

// 订阅 canal
canilClient.subscribe("myapp.orders", event -> {
    if (event.getType() == EventType.UPDATE || event.getType() == EventType.INSERT) {
        Order order = orderMapper.selectById(event.getId());
        elasticsearchOperations.save(order);
    } else {
        elasticsearchOperations.delete(event.getId().toString(), Order.class);
    }
});

5.2 DB → MongoDB(异步日志)

@KafkaListener(topics = "operation-logs")
public void handleLog(OperationLog log) {
    mongoTemplate.insert(log);   // 存到 MongoDB
}

六、限流 + 降级 + 熔断

6.1 完整方案

6.2 Resilience4j 熔断

@CircuitBreaker(name = "inventory", fallbackMethod = "fallback")
public InventoryResponse getInventory(Long productId) {
    return inventoryClient.getById(productId);
}

private InventoryResponse fallback(Long productId, Throwable t) {
    log.warn("库存服务降级", t);
    return new InventoryResponse(productId, "未知", 0);
}

七、分布式配置中心

# Nacos Config
spring:
  cloud:
    nacos:
      config:
        server-addr: localhost:8848
        file-extension: yaml

八、监控大盘

九、典型中间件组合

场景	组合
电商后台	MySQL + Redis + Kafka + ES
内容平台	MySQL + ES + MongoDB(评论) + OSS
金融系统	MySQL + Redis + RocketMQ
物联网	TDengine / InfluxDB + Kafka
社交	MySQL + Redis + MongoDB + ES

十、容量评估

# 示例:日均 100 万订单
QPS: 100w / 86400 ≈ 12 QPS(平均),峰值 ≈ 120 QPS
存储: 100w * 1KB = 1GB/天 → 365GB/年
缓存: 热点 1% → 1w key,Redis 内存 1GB 够用

十一、本章小结

中间件	主要职责
Redis	缓存、限流、锁
MySQL	主业务数据
MongoDB	文档 / 日志
Kafka / RabbitMQ	异步消息
ES	搜索 / 分析

动手练习

1. 整合 Redis + MySQL + Kafka 实现订单
2. 实现多级缓存(本地 + Redis + DB)
3. 用 Canal 同步 MySQL → ES
4. 配置 Prometheus + Grafana 监控

推荐阅读

• 📖 中间件综合架构

---

下一章:第 206 章:Linux 基础与常用命令 →