TCP服务器无法注册第二个传入流问题求助

架构老林

2026-7-18

TCP服务器无法注册第二个传入连接流排查与修复

我正在开发一款客户端-服务器应用，服务器负责处理连接、注册用户，并实现用户线程间的消息转发。目前遇到TCP服务器无法注册第二个传入流的问题，尝试过克隆TcpStream、添加服务器响应等方案，但均未解决问题，恳请帮忙排查原因。

服务器代码

use std::collections::HashMap;
use std::hash::Hash;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::Component::ParentDir;
use std::string::ToString;
use std::sync::{Arc, mpsc, Mutex};
use std::sync::mpsc::Receiver;
use std::thread;
use std::thread::spawn;

const GET_REQ: &str = "GET / HTTP/1.1\r\n";

fn spawn_thread(mut stream: TcpStream, uuid: String, reciver: Arc<Mutex<Receiver<String>>>)
{
    thread::spawn(move || {
        let reciver = reciver.lock().unwrap();
        let userID = uuid;
        println!("created new user: {userID}");
        loop {
            match reciver.recv() {
                Ok(message) => {
                    println!("{message}");
                    let splitted_message = split_message(&message);
                    println!("{:?}", splitted_message);
                    let end_user = splitted_message.get(2);
                    match end_user {
                        Some(end_user) => {
                            print!("tf ");
                            if end_user == &userID {
                                println!("woah");
                                stream.write(splitted_message.get(1).unwrap().as_bytes()).unwrap();
                            }
                        }
                        None => { println!("frick"); }
                    }
                }
                Err(_) => {
                    println!("wtf");
                    continue;
                }
            }
            print!("a");
        }
    });
}
fn split_message(messaggio: &String) -> Vec<String>{
    messaggio.split("\n").map(|x| x.to_string()).collect::<Vec<String>>()
}
fn person_already_connected(uuid: String, connected_ppl: &HashMap<String, bool>) -> bool{
    connected_ppl.get(&*uuid).is_some()
}
fn get_uuid(buffer: &String) -> String{
    let mut parameters: String = buffer.split("\r\n").filter(|x| x.contains("X-user-id")).collect();
    parameters = parameters.split(" : ").nth(1).unwrap().to_string();
    parameters
}
fn get_buffer_from_stream(stream: &mut TcpStream) -> String{
    let mut buffer = [0; 1024];
    stream.read(&mut buffer).unwrap();
    let buffer: String = String::from_utf8_lossy(&buffer).to_string();
    buffer
}
fn add_person_to_map(uuid: String, mappa: &mut HashMap<String, bool>){
    mappa.insert(uuid, true);
}

fn main (){
    let mut listener = TcpListener::bind("0.0.0.0:7878").unwrap();
    // listener.set_nonblocking(true).expect("it blocks");
    let ( mut sender, mut reciver) = mpsc::channel::<String>();
    let mut reciver = Arc::new(Mutex::new(reciver));
    let mut connected: HashMap<String, bool> = HashMap::new();
    for stream in listener.incoming(){
        match stream {
            Ok(mut stream) => {
                println!("recived stream");
                let buffer = get_buffer_from_stream(&mut stream);
                println!("buffer: {:?}", buffer);
                if buffer.starts_with(GET_REQ) && !person_already_connected(get_uuid(&buffer), &connected) {
                    add_person_to_map(get_uuid(&buffer), &mut connected);
                    let reciver = Arc::clone(&reciver);
                    spawn_thread(stream, get_uuid(&buffer), reciver);
                }
                else {
                    println!("sent stream");
                    sender.send(buffer).unwrap();
                }
            }
            _ => {continue}
        }

    }
}

客户端代码

use std::fmt::format;
use std::net::{TcpStream};
use std::io::{Read, Write};
use std::str::from_utf8;
use std::thread;
use std::time::Duration;
use uuid::Uuid;

fn main() {
    let uuid = Uuid::new_v4();
    let get_req =
        format!("GET / HTTP/1.1\r\nHost: yeahyeah\r\nAccept: */*\r\nX-user-id : {uuid}\r\n");

    match TcpStream::connect("myIp:7878") {
        Ok(mut stream) => {
            println!("Successfully connected to server in port 7878");

            let msg = format!("{uuid}\nplease work please please\n{uuid}");

            stream.write(get_req.as_bytes()).unwrap();
            thread::sleep(Duration::from_millis(100));
            println!("after the sleep");
            stream.write(msg.as_bytes()).unwrap();
            println!("Sent Hello, awaiting reply...");

            let mut data = [0 as u8; 1024]; // using 6 byte buffer
            match stream.read(&mut data) {
                Ok(_) => {
                    println!("text recived: {}", String::from_utf8_lossy(&data))
                },
                Err(e) => {
                    println!("Failed to receive data: {}", e);
                }
            }
        },
        Err(e) => {
            println!("Failed to connect: {}", e);
        }
    }
    println!("Terminated.");
}

核心问题分析

1. 全局Receiver锁被永久持有

spawn_thread中，线程启动时就获取reciver.lock().unwrap()且全程不释放，导致第一个用户线程独占锁，后续所有线程无法获取锁而阻塞，完全无法处理第二个用户的注册。

2. MPSC通道设计错误

标准库MPSC的Receiver是独占的，只能有一个线程接收消息，用Arc共享的方式根本无法实现多线程消息分发，这是设计上的根本性错误。

3. 无同步保护的HashMap操作

connected HashMap在主线程中被修改，没有任何同步机制，一旦多线程并发操作会引发数据竞争，导致未定义行为。

4. TCP读取与解析的脆弱性

get_buffer_from_stream用read.unwrap()，不仅可能读取不完整请求，还会在读取失败时直接panic导致服务器崩溃。
get_uuid中多次使用unwrap，一旦请求格式错误就会panic，且重复解析buffer效率低下。

修复方案

1. 替换为广播通道

使用crossbeam-channel的广播功能实现多线程消息分发，先在Cargo.toml添加依赖：

crossbeam-channel = "0.5"

2. 重构服务器代码

use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::thread;
use crossbeam_channel::{unbounded, Receiver, Sender};

const GET_REQ: &str = "GET / HTTP/1.1\r\n";

fn spawn_thread(mut stream: TcpStream, user_id: String, rx: Receiver<String>, global_tx: Arc<Mutex<Sender<String>>>) {
    thread::spawn(move || {
        println!("created new user: {user_id}");
        let mut buffer = [0; 1024];
        
        loop {
            // 同时监听广播消息和用户输入
            crossbeam_channel::select! {
                recv(rx) -> msg => {
                    match msg {
                        Ok(message) => {
                            let splitted_message = split_message(&message);
                            if let Some(end_user) = splitted_message.get(2) {
                                if end_user == &user_id {
                                    if let Some(content) = splitted_message.get(1) {
                                        if stream.write(content.as_bytes()).is_err() {
                                            println!("User {user_id} disconnected");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                        Err(_) => break,
                    }
                },
                default => {
                    // 读取用户发送的消息并广播
                    match stream.read(&mut buffer) {
                        Ok(0) => {
                            println!("User {user_id} disconnected");
                            break;
                        }
                        Ok(n) => {
                            let msg = String::from_utf8_lossy(&buffer[..n]).trim().to_string();
                            if global_tx.lock().unwrap().send(msg).is_err() {
                                println!("Failed to broadcast message");
                            }
                        }
                        Err(e) => {
                            println!("User {user_id} read error: {}", e);
                            break;
                        }
                    }
                }
            }
        }
    });
}

fn split_message(messaggio: &str) -> Vec<String> {
    messaggio.split("\n").map(|x| x.trim().to_string()).collect()
}

fn get_uuid(buffer: &str) -> Option<String> {
    buffer.split("\r\n")
        .find(|line| line.contains("X-user-id"))
        .and_then(|line| line.split(" : ").nth(1))
        .map(|s| s.trim().to_string())
}

fn get_buffer_from_stream(stream: &mut TcpStream) -> String {
    let mut buffer = [0; 1024];
    match stream.read(&mut buffer) {
        Ok(n) => String::from_utf8_lossy(&buffer[..n]).to_string(),
        Err(_) => String::new(),
    }
}

fn main() {
    let listener = TcpListener::bind("0.0.0.0:7878").unwrap();
    let (tx, _) = unbounded::<String>();
    let global_tx = Arc::new(Mutex::new(tx));
    let connected = Arc::new(Mutex::new(HashMap::<String, bool>::new()));

    for stream in listener.incoming() {
        match stream {
            Ok(mut stream) => {
                println!("received stream");
                let buffer = get_buffer_from_stream(&mut stream);
                println!("buffer: {:?}", buffer);

                if buffer.starts_with(GET_REQ) {
                    if let Some(uuid) = get_uuid(&buffer) {
                        let mut connected_lock = connected.lock().unwrap();
                        if !connected_lock.contains_key(&uuid) {
                            connected_lock.insert(uuid.clone(), true);
                            let rx = global_tx.lock().unwrap().subscribe();
                            spawn_thread(stream, uuid, rx, Arc::clone(&global_tx));
                        } else {
                            println!("User {uuid} already connected");
                        }
                    } else {
                        println!("Invalid X-user-id in request");
                    }
                } else {
                    if !buffer.is_empty() {
                        if global_tx.lock().unwrap().send(buffer).is_err() {
                            println!("Failed to broadcast message");
                        }
                    }
                }
            }
            Err(e) => println!("Connection error: {}", e),
        }
    }
}

3. 优化客户端代码

修复HTTP请求格式（添加结束符\r\n\r\n），移除不必要的sleep，增强错误处理：

use std::net::TcpStream;
use std::io::{Read, Write};
use uuid::Uuid;

fn main() {
    let uuid = Uuid::new_v4().to_string();
    // HTTP请求必须以\r\n\r\n结尾，否则服务器无法识别请求结束
    let get_req = format!(
        "GET / HTTP/1.1\r\nHost: yeahyeah\r\nAccept: */*\r\nX-user-id : {uuid}\r\n\r\n"
    );

    match TcpStream::connect("127.0.0.1:7878") {
        Ok(mut stream) => {
            println!("Successfully connected to server in port 7878");

            let msg = format!("{uuid}\nplease work please please\n{uuid}");

            if stream.write(get_req.as_bytes()).is_err() {
                println!("Failed to send registration request");
                return;
            }

            if stream.write(msg.as_bytes()).is_err() {
                println!("Failed to send message");
                return;
            }
            println!("Sent message, awaiting reply...");

            let mut data = [0 as u8; 1024];
            match stream.read(&mut data) {
                Ok(n) if n > 0 => {
                    println!("text received: {}", String::from_utf8_lossy(&data[..n]))
                }
                Ok(_) => println!("Server closed connection"),
                Err(e) => println!("Failed to receive data: {}", e),
            }
        }
        Err(e) => println!("Failed to connect: {}", e),
    }
    println!("Terminated.");
}

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

TCP服务器无法注册第二个传入流问题求助

服务器代码 #

客户端代码 #

核心问题分析 #

1. 全局Receiver锁被永久持有 #

2. MPSC通道设计错误 #

3. 无同步保护的HashMap操作 #

4. TCP读取与解析的脆弱性 #

修复方案 #

1. 替换为广播通道 #

2. 重构服务器代码 #

3. 优化客户端代码 #