C++ 分布式语音识别服务实践

C++ 分布式语音识别服务实践 | 极客日志

C++ 分布式语音识别服务实践

一、项目背景与核心功能

基于 C++ 实现了一个分布式语音识别子服务，核心目标是提供高可用的 RPC 接口，支持客户端上传 PCM 音频文件并返回识别结果。技术栈选型如下：

RPC 框架：brpc（百度开源高性能 RPC 框架，支持多种协议）；
数据序列化：Protobuf（定义 RPC 接口和数据结构）；
服务注册与发现：etcd（分布式键值存储，实现服务上下线感知）；
语音识别能力：百度 AI 语音 SDK（提供成熟的 PCM 音频转文字能力）；
日志与配置：spdlog（高性能日志库）、gflags（命令行参数解析）。

项目分为服务端和客户端两部分：

服务端：实现 RPC 服务、注册到 etcd、封装百度 AI SDK 调用；
客户端：通过 etcd 发现服务、读取音频文件、发起 RPC 请求。

二、核心代码架构解析

为了保证代码的可扩展性和可维护性，采用'模块化 + Builder 模式'设计，各组件职责单一，解耦清晰。

1. 整体架构概览

语音识别服务
├─ 服务端（speech_server）
│  ├─ RPC 服务实现（SpeechServiceImpl）：处理语音识别请求
│  ├─ 服务构建器（SpeechServerBuilder）：组装各模块（ASR、注册、RPC）
│  ├─ 语音识别封装（ASRClient）：调用百度 AI SDK
│  ├─ 服务注册（Registry）：将服务节点注册到 etcd
│  └─ 日志配置：初始化 spdlog 日志
└─ 客户端（speech_client）
   ├─ 服务发现（Discovery）：从 etcd 获取服务节点
   ├─ 信道管理（ServiceManager）：RR 轮询负载均衡
   └─ 音频读取：调用百度 AI SDK 工具函数读取 PCM 文件

2. 关键模块代码解析

（1）RPC 接口定义（Protobuf）

首先通过 speech.proto 定义 RPC 服务和数据结构，明确请求（音频数据）和响应（识别结果）格式：

syntax = "proto3";
package zrt; // 命名空间，避免类名冲突
option cc_generic_services = true; // 生成 C++ RPC 服务代码

// 语音识别请求
message SpeechRecognitionReq {
    string request_id = 1; // 请求 ID（用于追踪）
    bytes speech_content = 2; // 核心：PCM 音频数据（二进制）
    optional string user_id = 3; // 可选：用户 ID
    optional string session_id = 4; // 可选：会话 ID（鉴权用）
}

// 语音识别响应
message SpeechRecognitionRsp {
    string request_id = 1; // 对应请求的 ID
    bool success = 2; // 识别是否成功
    optional string errmsg = 3; // 失败原因（success=false 时必选）
    optional string recognition_result = 4; // 识别结果（success=true 时必选）
}

// RPC 服务定义
service SpeechService {
    rpc SpeechRecognition(SpeechRecognitionReq) returns (SpeechRecognitionRsp);
}

通过 protoc 编译生成 speech.pb.cc 和 speech.pb.h，为 RPC 服务提供基础代码。

（2）语音识别封装（ASRClient）

封装百度 AI SDK 的调用逻辑，对外提供简洁的 recognize 接口，隐藏 SDK 细节：

#pragma once
#include "../third/include/aip-cpp-sdk/speech.h"
#include "logger.hpp"

namespace zrt {
class ASRClient {
public:
    using ptr = std::shared_ptr<ASRClient>;
    
    // 初始化：传入百度 AI 的 AppID、APIKey、SecretKey
    ASRClient(const std::string &app_id, const std::string &api_key, const std::string &secret_key)
        : _client(app_id, api_key, secret_key) {}
    
    // 核心接口：输入 PCM 音频数据，输出识别结果
    std::string recognize(const std::string &speech_data, std::string &err) {
        // 调用百度 SDK：PCM 格式（16k 采样率）
        Json::Value result = _client.recognize(speech_data, "pcm", 16000, aip::null);
        
        // 处理 SDK 返回：err_no=0 表示成功
        if (result["err_no"].asInt() != 0) {
            LOG_ERROR("语音识别失败：{}", result["err_msg"].asString());
            err = result["err_msg"].asString();
            return "";
        }
        return result["result"][0].asString(); // 返回第一个识别结果
    }

private:
    aip::Speech _client; // 百度 AI SDK 的 Speech 客户端
};
}

（3）RPC 服务实现（SpeechServiceImpl）

继承 Protobuf 生成的服务基类，实现 SpeechRecognition 接口，处理客户端请求：

class SpeechServiceImpl : public zrt::SpeechService {
public:
    // 注入 ASRClient 实例（依赖注入，解耦服务与 ASR 实现）
    SpeechServiceImpl(const ASRClient::ptr &asr_client) : _asr_client(asr_client) {}
    
    void SpeechRecognition(google::protobuf::RpcController* controller,
                           const ::zrt::SpeechRecognitionReq* request,
                           ::zrt::SpeechRecognitionRsp* response,
                           ::google::protobuf::Closure* done) {
        LOG_DEBUG("收到语音转文字请求！request_id: {}", request->request_id());
        brpc::ClosureGuard rpc_guard(done); // 自动释放 Closure，避免内存泄漏
        
        // 1. 调用 ASRClient 识别音频
        std::string err;
        std::string res = _asr_client->recognize(request->speech_content(), err);
        
        // 2. 组装响应
        response->set_request_id(request->request_id());
        if (res.empty()) {
            // 识别失败：设置错误信息
            response->set_success(false);
            response->set_errmsg("语音识别失败:" + err);
            return;
        }
        // 识别成功：返回结果
        response->set_success(true);
        response->set_recognition_result(res);
    }

private:
    ASRClient::ptr _asr_client; // 语音识别客户端
};

（4）服务注册与发现（etcd 集成）

服务注册（Registry）：将服务节点注册到 etcd，并通过 lease（租约）维持节点存活，服务下线时自动删除；
服务发现（Discovery）：监听 etcd 的服务目录，感知服务上下线，并回调更新信道。

以服务发现为例，核心代码：

class Discovery {
public:
    using ptr = std::shared_ptr<Discovery>;
    using NotifyCallback = std::function<void(std::string, std::string)>;
    
    // 初始化：连接 etcd、拉取现有服务、监听变化
    Discovery(const std::string &host, const std::string &basedir, 
              const NotifyCallback &put_cb, const NotifyCallback &del_cb)
        : _client(std::make_shared<etcd::Client>(host)), _put_cb(put_cb), _del_cb(del_cb) {
        
        // 1. 拉取现有服务节点（服务启动时初始化）
        auto resp = _client->ls(basedir).get();
        if (!resp.is_ok()) {
            LOG_ERROR("获取服务列表失败：{}", resp.error_message());
            return;
        }
        
        // 遍历现有节点，调用上线回调
        for (int i = 0; i < resp.keys().size(); ++i) {
            if (_put_cb) _put_cb(resp.key(i), resp.value(i).as_string());
        }
        
        // 2. 监听 etcd 目录变化（实时感知上下线）
        _watcher = std::make_shared<etcd::Watcher>(
            *_client.get(), basedir, 
            std::bind(&Discovery::callback, this, std::placeholders::_1), 
            true // 递归监听子目录
        );
    }

private:
    // etcd 事件回调：处理 PUT（上线）和 DELETE（下线）事件
      {
         (!resp.()) {
            (, resp.());
            ;
        }
         ( &ev : resp.()) {
             (ev.() == etcd::Event::PUT) {
                (, ev.().(), ev.().());
                 (_put_cb) _put_cb(ev.().(), ev.().());
            }   (ev.() == etcd::Event::DELETE_) {
                (, ev.().(), ev.().());
                 (_del_cb) _del_cb(ev.().(), ev.().());
            }
        }
    }

:
    NotifyCallback _put_cb; 
    NotifyCallback _del_cb; 
    std::shared_ptr<etcd::Client> _client; 
    std::shared_ptr<etcd::Watcher> _watcher; 
};

（5）信道管理与负载均衡（ServiceManager）

客户端通过 ServiceManager 管理 RPC 信道，采用 RR（Round-Robin）轮询策略实现负载均衡，避免单节点压力过大：

class ServiceManager {
public:
    using ptr = std::shared_ptr<ServiceManager>;
    
    // 声明需要关注的服务（只处理声明过的服务）
    void declared(const std::string &service_name) {
        std::unique_lock<std::mutex> lock(_mutex);
        _follow_services.insert(service_name);
    }
    
    // 服务上线回调：添加信道
    void onServiceOnline(const std::string &service_instance, const std::string &host) {
        std::string service_name = getServiceName(service_instance);
        if (_follow_services.count(service_name) == 0) {
            LOG_DEBUG("{}服务上线，无需关注", service_name);
            return;
        }
        auto service = getOrCreateServiceChannel(service_name);
        service->append(host);
    }
    
    // 选择一个信道（RR 轮询）
    ServiceChannel::ChannelPtr choose(const std::string &service_name) {
        std::unique_lock<std::mutex> lock(_mutex);
        auto it = _services.find(service_name);
        if (it == _services.end()) {
            LOG_ERROR("无{}服务的可用节点", service_name);
            return nullptr;
        }
         it->second->(); 
    }

:
    {
         pos = service_instance.();
         pos == std::string::npos ? service_instance : service_instance.(, pos);
    }

:
    std::mutex _mutex;
    std::unordered_set<std::string> _follow_services;
    std::unordered_map<std::string, ServiceChannel::ptr> _services;
};

三、核心问题与解决方案（踩坑复盘）

在项目开发过程中，遇到了多个编译期和运行期问题，以下是关键问题的排查过程和解决方案，均为 C++ 分布式服务开发中的常见坑。

1. 编译期：百度 AI SDK 的 `toupper` 重载歧义

问题现象

编译客户端时，报 std::transform 调用 toupper 的重载歧义错误：

error: no matching function for call to 'transform(..., <unresolved overloaded function type>)'
note: couldn't deduce template parameter '_UnaryOperation'

原因分析

C++ 中有两个 toupper 版本，编译器无法确定使用哪个：

<cctype> 中的 int toupper(int c)：处理单个字符，参数为 int（兼容 EOF）；
<locale> 中的 template <class charT> charT toupper(charT c, const locale& loc)：带本地化参数的模板函数。

百度 AI SDK 的 utils.h 中直接调用 std::transform(..., toupper)，未明确版本，导致歧义。

解决方案

用 lambda 表达式显式指定 toupper 版本，消除歧义，并处理 char 类型转换（避免负数问题）：

// 修改前（SDK 原代码，错误）
std::transform(src.begin(), src.end(), src.begin(), toupper);

// 修改后（正确）
std::transform(src.begin(), src.end(), src.begin(), [](unsigned char c) {
    return static_cast<char>(std::toupper(c)); // 显式调用<cctype>版本
});

关键思路：lambda 作为'中间层'，明确参数类型和函数版本，让编译器无需猜测。

2. 编译期：函数漏写 `return` 语句的警告

问题现象

修改 utils.h 后，编译报'无返回语句'警告：

warning: no return statement in function returning non-void [-Wreturn-type]

原因分析

to_upper/to_lower 函数声明返回 std::string，但修改时不小心删除了 return src; 语句，导致函数无返回值（C++ 中属于未定义行为，编译器宽容处理为警告，但运行时可能返回随机值）。

解决方案

补全 return 语句，确保函数返回处理后的字符串：

std::string aip::to_upper(std::string src) {
    std::transform(...); // 处理逻辑
    return src; // 补全返回语句
}

3. 运行期：音频文件读取失败（`invalid audio length`）

问题现象

客户端运行时，输出 file_content.size() = 0，百度 AI SDK 返回'invalid audio length'：

0 语音识别失败:invalid audio length

原因分析

路径错误：客户端用相对路径 "16k.pcm"，但运行目录（如 build/）下无此文件；
文件权限：文件存在但无读权限；
格式错误：文件不是百度 SDK 要求的'16kHz 采样率、16 位深度、单声道'PCM。

解决方案

使用绝对路径：明确指定文件位置，避免相对路径陷阱：

// 修改前
aip::get_file_content("16k.pcm", &file_content);

// 修改后（替换为实际路径）
aip::get_file_content("/home/zrt/workspace/16k.pcm", &file_content);

验证文件权限：

# 查看权限，确保有 r（读）权限
ls -l 16k.pcm
# 无权限则添加
chmod +r 16k.pcm

验证 PCM 格式：用 ffmpeg 转换为标准格式：

# 将任意音频转为 16k、16 位、单声道 PCM
ffmpeg -i test.wav -ar 16000 -ac 1 -sample_fmt s16le 16k.pcm

4. 运行期：etcd `watcher` 警告（`watcher doesn't exit normally`）

问题现象

程序退出时，报 watcher doesn't exit normally 警告。

原因分析

Discovery 的 watcher 线程未正常停止，程序退出时强制终止线程导致警告。

解决方案

在 Discovery 析构函数中主动取消 watcher：

~Discovery() {
    _watcher->Cancel(); // 主动停止监听器
}

四、项目运行流程

服务端启动：

# 运行服务端（注册服务到 etcd）
./speech_server --etcd_host=http://127.0.0.1:2379 --speech_service=/service/speech_service

客户端调用：

# 运行客户端（发现服务并发起请求）
./speech_client --etcd_host=http://127.0.0.1:2379 --speech_service=/service/speech_service

成功输出：

12345 # file_content.size()，非 0 表示读取成功
收到响应：111111
收到响应：你好，世界

五、总结与经验

第三方 SDK 踩坑：第三方库代码可能不严谨（如百度 SDK 的 toupper 歧义），需针对性修改，修改时注意保留原功能；
分布式服务核心：服务注册发现（etcd）和负载均衡（RR）是分布式服务的基石，需保证高可用和线程安全；
C++ 编译问题：编译错误需重点看 error: 前的具体代码行，尤其是模板推导失败（如重载歧义），可通过显式类型或 lambda 解决；
路径与权限：文件操作尽量用绝对路径，避免运行目录依赖；权限问题在 Linux 下容易被忽略，需提前验证。

这个项目不仅实现了语音识别的核心功能，更重要的是梳理了 C++ 分布式服务的开发流程和问题排查思路，后续可扩展多节点部署、熔断降级等高级特性。

C++ 分布式语音识别服务实践