分布式微服务系统架构第103集：高性能落地实践，Kafka，分布式锁，缓存双写架构

加群联系作者vx：xiaoda0423

仓库地址：https://webvueblog.github.io/JavaPlusDoc/

https://1024bat.cn/

JVM 维度调优说明

优化点说明：

⚙️ JVM 层面性能优化说明：

使用特点总结

精简优化点说明

AppListener 用于系统启动后的初始化工作

// 启动 TCP 服务监听线程
new Thread(() -> {
    try {
        Server.bind(serverConfig.getPort());
        logger.info("TCP Server started on port {}", serverConfig.getPort());
    } catch (Exception e) {
        logger.error("Server 启动失败", e);
    }
}, "Server-Thread").start();

调优与说明

• ✅ 可用线程池（如 Spring @Async + 配置线程池）替代裸线程。

• ✅ 将线程改为 守护线程（setDaemon(true)）可选优化。

• ✅ 后期支持优雅关闭，可以整合 Spring SmartLifecycle。

配置线程池（推荐）

@Configuration
@EnableAsync
public class AsyncConfig {
    @Bean("customTaskExecutor")
    public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(4);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("startup-task-");
        executor.initialize();
        return executor;
    }
}

✅ Thread.currentThread().isInterrupted()

检查当前线程是否已被中断（非清除标志位的查询）。

• 返回 true：线程被中断。

• 返回 false：线程未被中断。

🧠 不会清除中断标志位，适合在循环中定期检查是否应退出：

while (!Thread.currentThread().isInterrupted()) {
    // do work
}

✅ Thread.currentThread().interrupt()

设置中断标志位为 true，告诉线程：“你该停下来了”。

通常不是自己调用自己，而是从外部线程对目标线程进行中断通知：

Thread t = new Thread(() -> {
    while (!Thread.currentThread().isInterrupted()) {
        // do something
    }
});
t.start();

// 某些时刻需要中断它
t.interrupt(); // 设置中断标志位为 true

❗和 Thread.interrupted() 的区别

⚠️ 中断不会立刻停止线程！

中断是一种“协商式停止”：

• 线程收到中断信号后，需要自行检查中断标志并优雅退出。

• 否则它会继续运行。

🔄 结合 sleep()、wait()、join()

如果线程在执行这些方法时被中断，会抛出 InterruptedException，同时中断标志会被清除，需要手动再次调用 interrupt()：

try {
    Thread.sleep(1000);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt(); // 保留中断标志
    log.warn("线程睡眠中被中断");
}

✅ 实用模板

public void runTask() {
    while (!Thread.currentThread().isInterrupted()) {
        try {
            // 任务逻辑
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 恢复中断状态
            break; // 或 return，退出循环
        }
    }
    log.info("任务已中断并退出");
}

Kafka + 分布式锁 + 缓存双写架构

🔧 架构概览图

Kafka -> Spring KafkaListener
             |
             v
    分布式锁（Redisson）
             |
             v
      数据库查询 + 缓存双写（防穿透 + 延迟双删等）
             |
             v
         ack/commit offset

✅ Kafka 消费 + Redisson 分布式锁 + 缓存双写

消息处理类（以订单为例）

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderKafkaConsumer {

    private final RedissonClient redissonClient;
    private final OrderService orderService;
    private final CacheService cacheService;
    private final BloomFilter<String> bloomFilter;

    private static final String LOCK_PREFIX = "lock:order:";

    @KafkaListener(topics = "order-events", containerFactory = "kafkaListenerContainerFactory")
    public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String orderId = record.key();
        String message = record.value();

        if (!bloomFilter.mightContain(orderId)) {
            log.warn("BloomFilter reject orderId={}", orderId);
            ack.acknowledge(); // 直接ack，避免重复消费
            return;
        }

        String lockKey = LOCK_PREFIX + orderId;
        RLock lock = redissonClient.getLock(lockKey);

        try {
            if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
                // 业务处理逻辑：缓存双写
                handleOrderMessage(orderId, message);
            } else {
                log.warn("Order {} is already being processed.", orderId);
            }
        } catch (Exception e) {
            log.error("Kafka消费失败: {}", e.getMessage(), e);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
            ack.acknowledge(); // 无论成功失败都提交offset，避免阻塞（失败后消息需通过日志+补偿机制处理）
        }
    }

    private void handleOrderMessage(String orderId, String message) {
        // 1. 查询数据库
        Order order = orderService.findById(orderId);
        if (order == null) {
            log.warn("Order not found: {}", orderId);
            return;
        }

        // 2. 缓存写入（双写）
        cacheService.saveOrderToCache(order);

        // 3. 异步延迟双删逻辑（可选）
        CompletableFuture.delayedExecutor(5, TimeUnit.SECONDS).execute(() -> {
            cacheService.deleteOrderCache(orderId);
        });

        log.info("Order processed and cached: {}", orderId);
    }
}

✅ 关键组件说明

Redisson 分布式锁配置

@Bean
public RedissonClient redissonClient() {
    Config config = new Config();
    config.useSingleServer().setAddress("redis://localhost:6379");
    return Redisson.create(config);
}

缓存双写（CacheService）

public interface CacheService {
    void saveOrderToCache(Order order);
    void deleteOrderCache(String orderId);
}

@Service
public class RedisCacheService implements CacheService {

    private final RedisTemplate<String, Object> redisTemplate;

    public void saveOrderToCache(Order order) {
        String key = "order:" + order.getId();
        redisTemplate.opsForValue().set(key, order, Duration.ofMinutes(10));
    }

    public void deleteOrderCache(String orderId) {
        String key = "order:" + orderId;
        redisTemplate.delete(key);
    }
}

✅ 可选增强