第25章分布式系统中的并发

课程目录第25章分布式系统中的并发

25、分布式系统中的并发：消息队列（RabbitMQ/Kafka）的C++客户端、分布式锁的实现

聊到分布式系统里的并发控制，很多人的第一反应是「加锁」。但单机上的 std::mutex 在分布式环境里根本不管用——进程都不在同一台机器上，你锁谁去？

我这些年做分布式中间件，踩过不少坑。说白了，分布式并发就两条路：要么用消息队列把并发请求串行化，要么用分布式锁来协调多个节点。今天咱们就聊聊这两条路在 C++ 里怎么落地。

25.1 消息队列：把并发变成顺序

先说说消息队列。你想想看，如果多个服务同时写数据库，冲突是难免的。但如果你让它们把请求都扔到队列里，然后由一个消费者串行处理——并发问题自然就解了。

我在项目中常用两种消息队列：RabbitMQ 和 Kafka。它们定位不同，选型时得看场景。

25.1.1 RabbitMQ C++ 客户端实战

RabbitMQ 适合任务分发、RPC 调用这类场景。C++ 这边我用的是 SimpleAmqpClient 或者 amqpcpp 库。下面是我常用的一个生产者示例：

#include <SimpleAmqpClient/SimpleAmqpClient.h>
#include <iostream>

using namespace AmqpClient;

int main() {
    // 连接到 RabbitMQ 服务器
    Channel::ptr_t channel = Channel::Create("localhost", 5672, "guest", "guest");
    
    // 声明一个队列（如果不存在则创建）
    std::string queue_name = "task_queue";
    channel->DeclareQueue(queue_name, false, true, false, false);
    
    // 发送消息
    std::string message = "Hello from C++ client";
    channel->BasicPublish("", queue_name, 
                          BasicMessage::Create(message), 
                          false, false);
    
    std::cout << " [x] Sent '" << message << "'" << std::endl;
    return 0;
}

消费者这边，我习惯用 BasicConsume 配合回调：

#include <SimpleAmqpClient/SimpleAmqpClient.h>
#include <iostream>

using namespace AmqpClient;

int main() {
    Channel::ptr_t channel = Channel::Create("localhost", 5672, "guest", "guest");
    
    std::string queue_name = "task_queue";
    channel->DeclareQueue(queue_name, false, true, false, false);
    
    // 设置每次只取一条消息（手动确认）
    channel->BasicQos(0, 1, false);
    
    // 开始消费
    std::string consumer_tag = channel->BasicConsume(queue_name, "", true, false, false, 1);
    
    std::cout << " [*] Waiting for messages. To exit press CTRL+C\n";
    
    while (true) {
        Envelope::ptr_t envelope = channel->BasicConsumeMessage(consumer_tag);
        std::string body = envelope->Message()->Body();
        
        std::cout << " [x] Received '" << body << "'" << std::endl;
        
        // 模拟处理耗时
        std::this_thread::sleep_for(std::chrono::seconds(1));
        
        // 手动确认消息
        channel->BasicAck(envelope);
    }
    
    return 0;
}

我的经验：RabbitMQ 的 BasicQos 一定要设。我曾经有个项目没设这个，结果消费者一次拉了几百条消息到内存，处理不过来直接 OOM 了。设成 BasicQos(0, 1, false) 表示每次只取一条，处理完再取下一条——这才是真正的串行化。

25.1.2 Kafka C++ 客户端实战

Kafka 适合高吞吐、日志收集、事件溯源。C++ 这边我用 librdkafka，它提供了 C++ 封装。生产者示例：

#include <librdkafka/rdkafkacpp.h>
#include <iostream>
#include <string>

int main() {
    std::string errstr;
    
    // 配置生产者
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("bootstrap.servers", "localhost:9092", errstr);
    conf->set("acks", "all", errstr);  // 等待所有副本确认
    
    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        return 1;
    }
    
    delete conf;
    
    std::string topic_name = "test_topic";
    std::string payload = "Hello Kafka from C++";
    
    // 发送消息
    RdKafka::ErrorCode err = producer->produce(
        topic_name,
        RdKafka::Topic::PARTITION_UA,  // 自动选择分区
        RdKafka::Producer::RK_MSG_COPY,
        const_cast<char*>(payload.c_str()),
        payload.size(),
        nullptr, 0,
        nullptr
    );
    
    if (err != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Produce failed: " << RdKafka::err2str(err) << std::endl;
    } else {
        std::cout << "Produced message: " << payload << std::endl;
    }
    
    // 刷新发送缓冲区
    producer->flush(10000);
    
    delete producer;
    return 0;
}

消费者这边，我通常用 RdKafka::KafkaConsumer：

#include <librdkafka/rdkafkacpp.h>
#include <iostream>
#include <vector>

int main() {
    std::string errstr;
    
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("bootstrap.servers", "localhost:9092", errstr);
    conf->set("group.id", "my_group", errstr);
    conf->set("enable.auto.commit", "true", errstr);
    conf->set("auto.offset.reset", "earliest", errstr);
    
    RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);
    if (!consumer) {
        std::cerr << "Failed to create consumer: " << errstr << std::endl;
        return 1;
    }
    
    delete conf;
    
    // 订阅主题
    std::vector<std::string> topics = {"test_topic"};
    RdKafka::ErrorCode err = consumer->subscribe(topics);
    if (err != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Subscribe failed: " << RdKafka::err2str(err) << std::endl;
        return 1;
    }
    
    std::cout << "Consuming messages..." << std::endl;
    
    while (true) {
        RdKafka::Message *msg = consumer->consume(1000);  // 超时1秒
        
        switch (msg->err()) {
            case RdKafka::ERR_NO_ERROR:
                std::cout << "Received: " 
                          << static_cast<const char*>(msg->payload()) 
                          << std::endl;
                break;
            case RdKafka::ERR__TIMED_OUT:
                break;
            default:
                std::cerr << "Error: " << msg->errstr() << std::endl;
        }
        
        delete msg;
    }
    
    consumer->close();
    delete consumer;
    return 0;
}

注意：Kafka 的 enable.auto.commit 默认是 true。我曾经在生产环境遇到过消费者处理到一半挂了，但偏移量已经自动提交了——重启后丢了一批消息。关键场景建议手动提交偏移量，处理完再 commit。

25.2 分布式锁：协调多个节点

消息队列能解决一部分并发问题，但有些场景你必须直接加锁。比如多个节点同时扣库存，你不能让它们都去队列里排队——延迟受不了。这时候就需要分布式锁。

分布式锁的核心要求就三条：互斥、防死锁、可重入。我常用的实现方式有两种：基于 Redis 和基于 ZooKeeper。

25.2.1 基于 Redis 的分布式锁

Redis 实现分布式锁，最经典的是 Redlock 算法。但说实话，我大部分场景用简单的 SET NX + 过期时间就够了。下面是我封装的一个 C++ 类：

#include <hiredis/hiredis.h>
#include <string>
#include <chrono>
#include <thread>

class DistributedLock {
public:
    DistributedLock(const std::string& redis_host, int redis_port)
        : host_(redis_host), port_(redis_port) {
        ctx_ = redisConnect(host_.c_str(), port_);
        if (ctx_ == nullptr || ctx_->err) {
            throw std::runtime_error("Redis connection failed");
        }
    }
    
    ~DistributedLock() {
        if (ctx_) redisFree(ctx_);
    }
    
    // 尝试获取锁，超时返回 false
    bool tryLock(const std::string& lock_key, 
                 const std::string& lock_value,
                 int ttl_seconds) {
        // SET key value NX EX seconds
        std::string cmd = "SET " + lock_key + " " + lock_value + 
                          " NX EX " + std::to_string(ttl_seconds);
        
        redisReply *reply = (redisReply*)redisCommand(ctx_, cmd.c_str());
        if (!reply) return false;
        
        bool acquired = (reply->type == REDIS_REPLY_STATUS &&
                        std::string(reply->str) == "OK");
        freeReplyObject(reply);
        return acquired;
    }
    
    // 释放锁（使用 Lua 脚本保证原子性）
    bool unlock(const std::string& lock_key, 
                const std::string& lock_value) {
        // Lua 脚本：只有 value 匹配才删除
        const char *lua_script = R"(
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            else
                return 0
            end
        )";
        
        redisReply *reply = (redisReply*)redisCommand(
            ctx_, "EVAL %s 1 %s %s", 
            lua_script, lock_key.c_str(), lock_value.c_str()
        );
        
        if (!reply) return false;
        bool released = (reply->integer == 1);
        freeReplyObject(reply);
        return released;
    }
    
private:
    std::string host_;
    int port_;
    redisContext *ctx_;
};

核心要点：释放锁时一定要用 Lua 脚本检查 value。为什么？因为锁可能因为超时被自动释放了，然后被别的线程获取。如果你直接 DEL，就把别人的锁给删了——这就是经典的「误删锁」问题。我当年就因为这个坑，导致线上库存数据乱了好几天。

25.2.2 基于 ZooKeeper 的分布式锁

ZooKeeper 实现分布式锁，原理是利用临时顺序节点。每个客户端创建一个临时顺序节点，序号最小的那个获得锁。其他客户端监听前一个节点的删除事件。

我用的是 zookeeper C 库的 C++ 封装。核心逻辑如下：

#include <zookeeper/zookeeper.h>
#include <string>
#include <vector>
#include <algorithm>

class ZKDistributedLock {
public:
    ZKDistributedLock(const std::string& hosts, int timeout_ms)
        : hosts_(hosts), timeout_ms_(timeout_ms) {
        // 连接 ZooKeeper
        zhandle_t *zh = zookeeper_init(hosts_.c_str(), 
                                       nullptr, timeout_ms_, 
                                       nullptr, nullptr, 0);
        if (!zh) {
            throw std::runtime_error("ZooKeeper init failed");
        }
        zh_ = zh;
    }
    
    ~ZKDistributedLock() {
        if (zh_) zookeeper_close(zh_);
    }
    
    // 尝试获取锁
    bool tryLock(const std::string& lock_path) {
        // 创建临时顺序节点
        std::string node_path = lock_path + "/lock-";
        char buffer[128];
        int ret = zoo_create(zh_, node_path.c_str(), 
                             nullptr, 0,
                             &ZOO_OPEN_ACL_UNSAFE,
                             ZOO_EPHEMERAL | ZOO_SEQUENCE,
                             buffer, sizeof(buffer));
        
        if (ret != ZOK) return false;
        
        my_node_ = buffer;
        
        // 获取所有子节点
        struct String_vector children;
        ret = zoo_get_children(zh_, lock_path.c_str(), 0, &children);
        if (ret != ZOK) return false;
        
        // 排序，找到最小的节点
        std::vector<std::string> nodes;
        for (int i = 0; i < children.count; ++i) {
            nodes.push_back(children.data[i]);
        }
        std::sort(nodes.begin(), nodes.end());
        
        // 如果我是最小的，获得锁
        std::string my_name = my_node_.substr(lock_path.length() + 1);
        bool acquired = (nodes[0] == my_name);
        
        deallocate_String_vector(&children);
        return acquired;
    }
    
    // 释放锁
    bool unlock() {
        if (my_node_.empty()) return false;
        int ret = zoo_delete(zh_, my_node_.c_str(), -1);
        my_node_.clear();
        return (ret == ZOK);
    }
    
private:
    std::string hosts_;
    int timeout_ms_;
    zhandle_t *zh_;
    std::string my_node_;
};

我的建议：ZooKeeper 锁的好处是天然支持「公平锁」——谁先创建节点谁先拿到锁。而且临时节点机制能自动处理客户端崩溃的情况，不会死锁。但缺点是性能不如 Redis，每次操作都要走 ZAB 协议。我一般只在需要强一致性的场景用 ZooKeeper，比如配置中心、选主。

25.3 消息队列 vs 分布式锁：怎么选？

很多同学问我：到底用消息队列还是分布式锁？我的回答是：看你的业务容忍什么。

维度	消息队列	分布式锁
核心思想	串行化请求	互斥访问
延迟	较高（入队出队有开销）	较低（直接操作）
吞吐量	高（可批量处理）	低（锁竞争严重）
一致性	最终一致	强一致
典型场景	订单处理、日志收集	库存扣减、选主
复杂度	中等	较高（需处理死锁、重入）

我个人的经验是：能上消息队列就别上分布式锁。消息队列天然解耦，出了问题也好排查。分布式锁一旦出问题——比如 Redis 挂了、网络分区了——整个系统都可能瘫痪。

但有些场景绕不开。比如秒杀系统的库存扣减，你让用户去排队等消息，体验太差了。这时候用 Redis 分布式锁，配合本地缓存，能做到毫秒级响应。

25.4 知识体系总览

下面这张图总结了本章的核心逻辑，我画的时候特意把「选型决策」放在了中间——因为这是实际项目中最重要的环节。

嗯，这张图把本章的知识结构串起来了。左边是消息队列的两大代表，右边是分布式锁的两种实现，中间是选型决策——说白了就是你在延迟和一致性之间怎么权衡。

最后说一句：不管用哪种方案，一定要做好监控和降级。我见过太多系统，分布式锁一挂，整个服务就瘫了。加个本地缓存做降级，虽然数据可能有点延迟，但至少服务不挂。

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

📚 课程目录

25、分布式系统中的并发：消息队列（RabbitMQ/Kafka）的C++客户端、分布式锁的实现

25.1 消息队列：把并发变成顺序

25.1.1 RabbitMQ C++ 客户端实战

25.1.2 Kafka C++ 客户端实战

25.2 分布式锁：协调多个节点

25.2.1 基于 Redis 的分布式锁

25.2.2 基于 ZooKeeper 的分布式锁

25.3 消息队列 vs 分布式锁：怎么选？

25.4 知识体系总览