第二十八章：多线程在分布式系统中的应用

分布式系统，说白了就是一群计算机一起干活。但怎么让它们协调一致？怎么让它们高效通信？这背后离不开多线程。今天我就聊聊RPC框架、消息队列和分布式锁这三个场景，看看多线程是怎么在分布式世界里大显身手的。

28.1 RPC框架中的线程模型

RPC（远程过程调用）是分布式系统的基石。你调用一个远程服务，感觉就像调用本地函数一样。但底层呢？线程模型决定了性能上限。

我参与过一个内部RPC框架的优化，当时线上服务经常超时。排查后发现，线程模型设计不合理是主因。

28.1.1 常见的RPC线程模型

RPC框架的线程模型，大致分三类：

模型	特点	适用场景
BIO + 线程池	一个连接一个线程，阻塞IO	连接数少，低并发
NIO + Reactor	事件驱动，少量线程处理大量连接	高并发，IO密集
异步 + Future/Promise	完全非阻塞，回调驱动	超高吞吐，延迟敏感

我个人习惯用NIO+Reactor模型。为什么呢？你想想看，一个服务端可能要处理上万连接，如果用BIO，线程数直接爆炸。线程切换的开销就能把CPU拖垮。

28.1.2 Reactor模型的线程分工

Reactor模型里，线程分角色：

• Acceptor线程：专门接受新连接。通常只有一个。
• IO线程（Reactor）：负责读写事件。一般2-4个。
• Worker线程池：处理业务逻辑。数量根据业务调整。

我在项目中遇到过一个问题：IO线程里做了耗时操作，导致新连接无法被接受。嗯，这里要注意：IO线程只做数据收发，千万别做业务处理。业务逻辑丢给Worker线程池去做。

// 伪代码：Reactor线程模型
void reactor_loop() {
    while (running) {
        // 非阻塞获取就绪事件
        events = selector.select(100); // 超时100ms
        for (event in events) {
            if (event.isAcceptable()) {
                // Acceptor线程处理新连接
                accept_connection(event.channel());
            } else if (event.isReadable()) {
                // IO线程读取数据
                byte[] data = read_data(event.channel());
                // 丢给Worker线程池处理
                worker_pool.submit(() -> process_request(data));
            }
        }
    }
}

28.2 消息队列并发消费

消息队列是分布式系统的「血管」。生产者发消息，消费者处理消息。但消费者怎么并发？这里坑不少。

28.2.1 单线程 vs 多线程消费

很多消息队列（比如Kafka）的消费者默认是单线程拉取。为什么？因为要保证分区内的消息顺序。

但单线程消费太慢了。怎么办？我常用的方案是：

• 方案一：增加分区数，每个分区一个消费者线程。
• 方案二：消费者内部用线程池处理消息。
• 方案三：批量拉取，批量处理。

我个人推荐方案二。为什么呢？因为分区数不能无限增加，而且分区多了，rebalance成本很高。内部线程池更灵活。

// 消息队列并发消费示例
void consume_messages() {
    while (true) {
        // 拉取一批消息
        List<Message> batch = poll(100); // 超时100ms
        if (batch.isEmpty()) continue;

        // 用线程池并发处理
        CountDownLatch latch = new CountDownLatch(batch.size());
        for (Message msg : batch) {
            thread_pool.submit(() -> {
                try {
                    process(msg);
                } finally {
                    latch.countDown();
                }
            });
        }
        // 等待所有消息处理完成
        latch.await(5, TimeUnit.SECONDS);
        // 提交偏移量
        commit(batch);
    }
}

28.2.2 消费幂等性

并发消费还有一个大坑：重复处理。消息队列保证「至少一次」投递，但不会保证「恰好一次」。所以消费者必须幂等。

我习惯的做法是：

• 每条消息带唯一ID
• 处理前先查数据库，看是否已处理
• 用分布式锁或数据库唯一索引防重

28.3 分布式锁实现

分布式锁，说白了就是让多台机器上的线程互斥访问共享资源。单机锁（比如mutex）只能管一个进程内的线程，跨机器就不行了。

28.3.1 基于Redis的分布式锁

Redis实现分布式锁，最经典。核心思路：

• 用SETNX命令尝试加锁
• 设置过期时间，防止死锁
• 释放锁时，检查是不是自己的锁

// Redis分布式锁示例
int try_lock(const char* key, const char* owner, int ttl_ms) {
    // SET key value NX PX ttl_ms
    // NX: 只有key不存在时才设置
    // PX: 设置过期时间
    redisReply* reply = redisCommand(conn,
        "SET %s %s NX PX %d", key, owner, ttl_ms);
    if (reply != NULL && reply->type == REDIS_REPLY_STATUS) {
        return 1; // 加锁成功
    }
    return 0; // 加锁失败
}

int unlock(const char* key, const char* owner) {
    // 用Lua脚本保证原子性
    const char* script =
        "if redis.call('get', KEYS[1]) == ARGV[1] then "
        "  return redis.call('del', KEYS[1]) "
        "else "
        "  return 0 "
        "end";
    redisReply* reply = redisCommand(conn,
        "EVAL %s 1 %s %s", script, key, owner);
    return reply->integer;
}

28.3.2 基于ZooKeeper的分布式锁

ZooKeeper的分布式锁，利用临时顺序节点实现。优点是可靠性高，缺点是性能不如Redis。

原理很简单：

• 在锁目录下创建临时顺序节点
• 检查自己是不是最小的节点
• 如果是，获得锁；如果不是，监听前一个节点

我参与过一个金融项目，要求锁必须绝对可靠，不能有「脑裂」问题。我们最终选了ZooKeeper。虽然慢一点，但安全第一。

28.3.3 分布式锁的常见问题

问题	原因	解决方案
锁超时	业务执行时间超过锁过期时间	锁续期（看门狗机制）
锁重入	同一个线程多次加锁	记录线程ID和重入次数
脑裂	Redis主从切换导致锁丢失	Redlock算法或ZooKeeper

28.4 本章知识体系

下面这张图，概括了多线程在分布式系统中的三个核心应用场景：

多线程在分布式系统中的应用，说白了就是解决三个问题：怎么通信（RPC）、怎么处理（消息队列）、怎么互斥（分布式锁）。每个问题都有成熟的方案，但具体选哪个，得看业务场景。

我个人建议：先理解原理，再动手实践。别一上来就搞复杂的框架。从简单的BIO模型开始，慢慢过渡到NIO、异步。分布式锁也是，先用Redis实现，遇到问题再升级到ZooKeeper。

---

公众号:蓝海资料掘金营，微信deep3321