如何解决C++多线程控制台输出混乱问题？lock_guard无效

原问题代码 #

#include <iostream>
#include <string>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <algorithm>
#include <cctype>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>

std::mutex io_mtx;
std::mutex mtx;
std::condition_variable cv;
std::queue<std::string> buffer;
int sockfd;

void sort_and_replace(std::string& str) {
    // 示例实现，根据实际需求调整
    std::sort(str.begin(), str.end());
    for (char& c : str) {
        if (c == '1') c = 'K';
        if (c == '3') c = 'B';
    }
}

size_t sum_of_digits(const std::string& str) {
    size_t sum = 0;
    for (char c : str) {
        if (std::isdigit(c)) sum += c - '0';
    }
    return sum;
}

void connect_to_server() {
    // 示例连接逻辑，根据实际需求调整
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    struct sockaddr_in serv_addr;
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(8080);
    inet_pton(AF_INET, "127.0.0.1", &serv_addr.sin_addr);
    
    while (connect(sockfd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) < 0) {
        std::lock_guard<std::mutex> lock(io_mtx);
        std::cerr << "Error connecting to program2, retrying... \n";
        sleep(1);
    }
    std::lock_guard<std::mutex> lock(io_mtx);
    std::cout << "Connected to server program2 \n";
}

void thread1() {
    std::string str;

    while (true) {
        {
            std::lock_guard<std::mutex> lock(io_mtx);
            std::cout << std::endl << "Enter line: ";
        }

        std::getline(std::cin, str);

        std::unique_lock<std::mutex> lock(mtx);
        if (str == "exit") break;

        if (str.size() > 64 || !std::all_of(str.begin(), str.end(), ::isdigit)) {
            {
                std::lock_guard<std::mutex> iolock(io_mtx);
                std::cerr << "String must include only numbers from 0 to 9 and less then 64 symbols\n";
            }
            continue;
        }

        // from my library(it sort and replace some exact symbols
        sort_and_replace(str);

        buffer.push(str);
        cv.notify_one();
    }
}

void thread2() {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [] { return !buffer.empty(); });

        std::string str = buffer.front();
        buffer.pop();

        {
            std::lock_guard<std::mutex> iolock(io_mtx);
            std::cout << "Processed line: " << str << std::endl;
        }

        size_t sum = sum_of_digits(str);
        std::string sum_str = std::to_string(sum);

        if (send(sockfd, sum_str.c_str(), sum_str.size(), 0) < 0) {
            {
                std::lock_guard<std::mutex> iolock(io_mtx);
                std::cerr << "Error.\n";
            }
            close(sockfd);
            connect_to_server();
        }
    }
}

int main() {
    connect_to_server();

    std::thread t1(thread1);
    std::thread t2(thread2);

    t1.join();
    t2.join();
    close(sockfd);
    return 0;
}

当前混乱输出 #

Error connecting to program2, retrying... 
Connected to server program2 
Enter line: 123 
Enter line: Processed line: 3KB1 
123 
Enter line: Processed line: 3KB1 
1234 
Enter line: Processed line: KB3KB1 
123 
Enter line: Processed line: 3KB1 
1234 
Enter line: Processed line: KB3KB1

期望输出 #

Enter line: 1234
Processed line: KB3KB1
Enter line: 123
Processed line: 3KB1

问题根源 #

当前代码仅用io_mtx保护了单个IO操作的原子性，但输入提示的显示与处理结果的输出之间没有同步机制：thread1输出"Enter line: "后立即释放锁，此时thread2可能抢占输出资源，导致"Enter line: "与"Processed line..."交错显示。同时，thread1在提交任务后会立刻进入下一轮循环显示输入提示，完全不等待thread2处理完成，进一步加剧了输出混乱。

修复后的代码 #

#include <iostream>
#include <string>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <algorithm>
#include <cctype>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>

std::mutex io_mtx;
std::mutex mtx;
std::condition_variable cv;
std::condition_variable cv_response; // 新增：用于同步输入与处理完成的信号
std::queue<std::string> buffer;
int sockfd;

void sort_and_replace(std::string& str) {
    std::sort(str.begin(), str.end());
    for (char& c : str) {
        if (c == '1') c = 'K';
        if (c == '3') c = 'B';
    }
}

size_t sum_of_digits(const std::string& str) {
    size_t sum = 0;
    for (char c : str) {
        if (std::isdigit(c)) sum += c - '0';
    }
    return sum;
}

void connect_to_server() {
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    struct sockaddr_in serv_addr;
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(8080);
    inet_pton(AF_INET, "127.0.0.1", &serv_addr.sin_addr);
    
    while (connect(sockfd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) < 0) {
        std::lock_guard<std::mutex> lock(io_mtx);
        std::cerr << "Error connecting to program2, retrying... \n";
        sleep(1);
    }
    std::lock_guard<std::mutex> lock(io_mtx);
    std::cout << "Connected to server program2 \n";
}

void thread1() {
    std::string str;
    bool first_run = true;

    while (true) {
        {
            std::lock_guard<std::mutex> lock(io_mtx);
            if (first_run) {
                std::cout << "Enter line: ";
                first_run = false;
            } else {
                // 仅在上一次处理完成后，才显示下一个输入提示
                std::cout << "\nEnter line: ";
            }
        }

        std::getline(std::cin, str);

        std::unique_lock<std::mutex> lock(mtx);
        if (str == "exit") break;

        if (str.size() > 64 || !std::all_of(str.begin(), str.end(), ::isdigit)) {
            {
                std::lock_guard<std::mutex> iolock(io_mtx);
                std::cerr << "String must include only numbers from 0 to 9 and less than 64 symbols\n";
            }
            continue;
        }

        sort_and_replace(str);

        buffer.push(str);
        cv.notify_one();
        // 等待thread2处理完当前任务，确保输出顺序
        cv_response.wait(lock, []{ return buffer.empty(); });
    }
}

void thread2() {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [] { return !buffer.empty(); });

        std::string str = buffer.front();
        buffer.pop();

        {
            std::lock_guard<std::mutex> iolock(io_mtx);
            std::cout << "Processed line: " << str << std::endl;
        }

        // 通知thread1可以开始下一轮输入
        cv_response.notify_one();

        size_t sum = sum_of_digits(str);
        std::string sum_str = std::to_string(sum);

        if (send(sockfd, sum_str.c_str(), sum_str.size(), 0) < 0) {
            {
                std::lock_guard<std::mutex> iolock(io_mtx);
                std::cerr << "Error.\n";
            }
            close(sockfd);
            connect_to_server();
        }
    }
}

int main() {
    connect_to_server();

    std::thread t1(thread1);
    std::thread t2(thread2);

    t1.join();
    t2.join();
    close(sockfd);
    return 0;
}

修复说明 #

1. 新增同步条件变量：添加cv_response，让thread1在提交任务后等待，直到thread2处理完当前任务并发出通知，确保"输入→处理→输出→下一次输入"的串行流程。
2. 调整输入提示逻辑：第一次运行直接显示输入提示，后续仅在上一次处理完成后才显示下一个提示，避免多余的空行和提前显示。
3. 保持IO操作原子性：保留io_mtx保护所有cout/cerr操作，防止单个IO操作被打断。

这样修改后，控制台输出会严格按照"输入提示→用户输入→处理结果输出→下一个输入提示"的顺序执行，完全符合期望的输出格式。

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