C++语音传输软件Socket循环编解码仅执行一次后阻塞的问题

我用C++开发语音传输软件，音频功能基于PortAudio和Opus实现，网络部分采用Boost Asio。处理呼叫的receiveCall与发起呼叫函数结构一致：先初始化依赖库，再通过Socket收发音频数据并完成编解码。但while(clientConnected)循环仅执行一次就发生阻塞，通过std::cout << dataSize仅打印一次可确认该问题。

问题代码 #

void Window::receiveCall(const std::string& ip) {

    // Initialize PortAudio and Opus
    PortAudioWrapper portAudio;
    OpusWrapper opus;
    if (!portAudio.OpenDefaultStream(1, 1, paFloat32, 48000, 960))
    {
        std::cerr << "Failed to open PortAudio stream" << std::endl;
        return;
    }
    if (!opus.Init(48000, 2, OPUS_APPLICATION_VOIP))
    {
        std::cerr << "Failed to initialize Opus" << std::endl;
        return;
    }
    if (!portAudio.StartStream())
    {
        std::cerr << "Failed to start PortAudio stream" << std::endl;
        return;
    }
    boost::asio::io_service ioService;
    boost::asio::ip::tcp::socket socket(ioService);
    boost::asio::ip::tcp::acceptor acceptor(ioService);

    try
    {
        // Create an acceptor to listen for incoming connections
        boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(ip), 12345);
        acceptor.open(endpoint.protocol());
        acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
        acceptor.bind(endpoint);
        acceptor.listen();

        std::cout << "Waiting for incoming connection on " << ip << ":12345" << std::endl;

        // Wait for a client to connect

        acceptor.accept(socket);

        std::cout << "Connected to client at " << socket.remote_endpoint().address().to_string() << std::endl;
        bool clientConnected = true;

        // Create a buffer to hold the audio data
        std::vector<unsigned char> buffer(1276);

        // Send and receive audio data
        while (clientConnected)
        {
            // Read audio data from PortAudio
            float audioData[960 * 2 * sizeof(float)];
            portAudio.readStream(audioData, 960);
            
            // Compress the audio data using Opus
            int dataSize = opus.Encode(audioData, 960, buffer.data(), 1276);
            std::cout << "dataSize : "<< dataSize << std::endl;
            // Send the compressed audio data to the client
            boost::asio::write(socket, boost::asio::buffer(buffer.data(), dataSize));

            // Receive compressed audio data from the client
            boost::system::error_code error;
            dataSize = socket.read_some(boost::asio::buffer(buffer), error);
            if (error == boost::asio::error::eof)
            {
                std::cout << "Client disconnected" << std::endl;
                clientConnected = false; // set flag to false when client disconnects
            }
            else if (error)
            {
                std::cerr << "Error receiving audio data: " << error.message() << std::endl;
                clientConnected = false; // set flag to false when client disconnects
            }

            // Decompress the audio data using Opus
            float decodedData[960 * 2 * sizeof(float)];
            int numSamples = opus.Decode(buffer.data(), dataSize, decodedData, 960);

            // Write the decompressed audio data to PortAudio
            portAudio.writeStream(decodedData, numSamples);
        }
    }    
    catch (std::exception& e) {
        std::cerr << "Exception in receiveCall: " << e.what() << std::endl;
    }
    // Cleanup code
    socket.close();
    acceptor.close();
    portAudio.StopStream();
}

问题分析 #

1. 同步IO互相阻塞：循环中先执行发送操作，随后调用socket.read_some()——这是同步阻塞接口，若对方未发送数据，当前线程会一直卡在该调用上，无法进入下一轮音频采集与发送流程，直接导致循环停滞。
2. 声道数不匹配：PortAudio初始化时设置的是单声道（OpenDefaultStream(1,1,...)），但Opus初始化用了2声道，虽不是阻塞直接原因，但会导致编解码逻辑异常，可能间接引发后续问题。
3. 音频缓冲区定义冗余：音频数据数组定义为960 * 2 * sizeof(float)，对应双声道，但实际是单声道采集，内存空间浪费且易引发数据处理错误。

解决方案 #

1. 拆分收发线程

将音频发送与接收逻辑拆分为两个独立线程，避免同步IO的互相阻塞：

#include <atomic>
#include <thread>

// 发送线程：负责采集、编码、发送音频
void sendAudio(PortAudioWrapper& portAudio, OpusWrapper& opus, boost::asio::ip::tcp::socket& socket, std::atomic<bool>& clientConnected) {
    std::vector<unsigned char> buffer(1276);
    // 单声道缓冲区，匹配PortAudio设置
    float audioData[960 * sizeof(float)];

    while (clientConnected) {
        portAudio.readStream(audioData, 960);
        int dataSize = opus.Encode(audioData, 960, buffer.data(), 1276);
        boost::asio::write(socket, boost::asio::buffer(buffer.data(), dataSize));
    }
}

// 接收线程：负责接收、解码、播放音频
void receiveAudio(PortAudioWrapper& portAudio, OpusWrapper& opus, boost::asio::ip::tcp::socket& socket, std::atomic<bool>& clientConnected) {
    std::vector<unsigned char> buffer(1276);
    float decodedData[960 * sizeof(float)];

    while (clientConnected) {
        boost::system::error_code error;
        int dataSize = socket.read_some(boost::asio::buffer(buffer), error);
        
        if (error) {
            std::cerr << "Receive error: " << error.message() << std::endl;
            clientConnected = false;
            break;
        }

        int numSamples = opus.Decode(buffer.data(), dataSize, decodedData, 960);
        portAudio.writeStream(decodedData, numSamples);
    }
}

// 修改后的receiveCall函数
void Window::receiveCall(const std::string& ip) {
    PortAudioWrapper portAudio;
    OpusWrapper opus;
    if (!portAudio.OpenDefaultStream(1, 1, paFloat32, 48000, 960)) {
        std::cerr << "Failed to open PortAudio stream" << std::endl;
        return;
    }
    // 修正Opus声道数，与PortAudio保持一致
    if (!opus.Init(48000, 1, OPUS_APPLICATION_VOIP)) {
        std::cerr << "Failed to initialize Opus" << std::endl;
        return;
    }
    if (!portAudio.StartStream()) {
        std::cerr << "Failed to start PortAudio stream" << std::endl;
        return;
    }

    boost::asio::io_service ioService;
    boost::asio::ip::tcp::socket socket(ioService);
    boost::asio::ip::tcp::acceptor acceptor(ioService);

    try {
        boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(ip), 12345);
        acceptor.open(endpoint.protocol());
        acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
        acceptor.bind(endpoint);
        acceptor.listen();

        std::cout << "Waiting for incoming connection on " << ip << ":12345" << std::endl;
        acceptor.accept(socket);
        std::cout << "Connected to client at " << socket.remote_endpoint().address().to_string() << std::endl;

        std::atomic<bool> clientConnected(true);
        std::thread sendThread(sendAudio, std::ref(portAudio), std::ref(opus), std::ref(socket), std::ref(clientConnected));
        std::thread recvThread(receiveAudio, std::ref(portAudio), std::ref(opus), std::ref(socket), std::ref(clientConnected));

        sendThread.join();
        recvThread.join();
    }
    catch (std::exception& e) {
        std::cerr << "Exception in receiveCall: " << e.what() << std::endl;
    }

    socket.close();
    acceptor.close();
    portAudio.StopStream();
}

2. 其他优化点

• 用std::atomic<bool>作为线程间的连接状态标志，保证线程安全；
• 修正音频缓冲区的大小，匹配单声道的采集与解码需求；
• 统一PortAudio与Opus的声道数参数，避免编解码逻辑异常。

内容的提问来源于stack exchange，提问作者Truitedepomme