10 - Kafka 异步架构#

┌────────────────── Kafka 在项目中的角色 ──────────────────┐
│                                                          │
│  ┌──────────────────┐      ┌──────────────────────┐     │
│  │   秒杀下单        │      │    点赞持久化          │     │
│  │                  │      │                      │     │
│  │  Lua成功         │      │  like()              │     │
│  │    ↓             │      │    ↓                  │     │
│  │  Producer ──→    │      │  Producer ──→        │     │
│  │  createOrder     │      │  TOPIC_LIKE_BEHAVIOR │     │
│  │    ↓             │      │    ↓                  │     │
│  │  Consumer        │      │  Consumer            │     │
│  │  → ThreadPool    │      │  → DB 写入            │     │
│  │  → Redisson 锁   │      │  → 批量计数聚合       │     │
│  │  → DB 写入       │      │                      │     │
│  │    ↓(失败)       │      │    ↓(失败)            │     │
│  │  save-order-     │      │  TOPIC_SAVE_DB_FAILED│     │
│  │  failed-topic    │      │                      │     │
│  │    ↓             │      │                      │     │
│  │  rollback.lua    │      │                      │     │
│  └──────────────────┘      └──────────────────────┘     │
└──────────────────────────────────────────────────────────┘

public class Event {
    private String topic;       // 目标 Topic
    private Long userId;        // 操作用户ID
    private Long entityId;      // 业务实体ID (orderId / behaviorId)
    private Map<String, Object> data;  // 扩展数据
}

// 秒杀示例:
Event {
    topic: "createOrder",
    userId: 1001,
    entityId: 7654321,  // orderId
    data: {
        "voucherId": 10,
        "buyNumber": 1
    }
}

@Component
public class KafkaOrderConsumer {
    // ⚠️ 自建线程池, 没有使用 Spring 的 ThreadPoolConfig Bean
    private final ExecutorService executorService =
            Executors.newFixedThreadPool(10);

    @KafkaListener(topics = {TOPIC_CREATE_ORDER})
    public void handleCreateOrder(ConsumerRecord record) {
        // 1. 解析消息
        Event event = JSONObject.parseObject(
                record.value().toString(), Event.class);

        // 2. 提交到线程池处理
        executorService.submit(() -> {
            VoucherOrder voucherOrder = new VoucherOrder()
                    .setId(event.getEntityId())
                    .setUserId(event.getUserId())
                    .setVoucherId(Long.valueOf(data.get("voucherId").toString()))
                    .setBuyNumber(Integer.valueOf(data.get("buyNumber").toString()));

            // 3. 调用 Service 处理 (内含 Redisson 锁 + DB 操作)
            voucherOrderService.createVoucherOrder(voucherOrder);
        });
        // ⚠️ 注意: 没有调用 acknowledgment.acknowledge()
    }

    @KafkaListener(topics = {TOPIC_SAVE_ORDER_FAILED})
    public void handleSaveOrderFailed(ConsumerRecord record) {
        // 执行 rollback.lua 恢复 Redis 数据
        Event event = JSONObject.parseObject(...);
        Long result = stringRedisTemplate.execute(
                ROLLBACK_SCRIPT,
                Collections.emptyList(),
                String.valueOf(event.getEntityId()),  // orderId
                voucherId.toString(),
                String.valueOf(buyNumber),
                event.getUserId().toString()
        );
    }

    @PreDestroy
    public void shutdown() {
        executorService.shutdown();
        if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
            executorService.shutdownNow();
        }
    }
}