Socket实现TCP双向通信

本文目录导读：

1. 基础双向通信示例
2. 增强版双向通信（带协议）
3. Python版双向通信
4. 测试示例
5. 关键特性和最佳实践

我来详细讲解如何使用Socket实现TCP双向通信,包括客户端和服务器端的实现。

基础双向通信示例

服务器端 (Node.js)

const net = require('net');
// 创建TCP服务器
const server = net.createServer((socket) => {
    console.log('客户端已连接');
    // 接收客户端数据
    socket.on('data', (data) => {
        const message = data.toString();
        console.log(`收到客户端消息: ${message}`);
        // 回复客户端
        socket.write(`服务器确认: 已收到 "${message}"`);
    });
    // 处理连接关闭
    socket.on('end', () => {
        console.log('客户端断开连接');
    });
    // 处理错误
    socket.on('error', (err) => {
        console.error('Socket错误:', err);
    });
    // 向客户端发送欢迎消息
    socket.write('欢迎连接到服务器!');
});
// 监听端口
server.listen(8080, () => {
    console.log('服务器启动在端口 8080');
});

客户端 (Node.js)

const net = require('net');
const readline = require('readline');
const client = new net.Socket();
// 创建命令行接口
const rl = readline.createInterface({
    input: process.stdin,
    output: process.stdout
});
// 连接到服务器
client.connect(8080, 'localhost', () => {
    console.log('已连接到服务器');
    // 启动用户输入循环
    console.log('请输入消息 (输入 "exit" 退出):');
    startUserInput();
});
// 接收服务器数据
client.on('data', (data) => {
    console.log(`\n服务器: ${data.toString()}`);
    console.log('请输入消息:');
});
// 处理连接关闭
client.on('close', () => {
    console.log('连接已关闭');
    process.exit(0);
});
// 处理错误
client.on('error', (err) => {
    console.error('错误:', err.message);
});
// 用户输入处理
function startUserInput() {
    rl.on('line', (input) => {
        if (input.toLowerCase() === 'exit') {
            console.log('断开连接...');
            client.end();
            rl.close();
            return;
        }
        // 发送消息到服务器
        client.write(input);
    });
}

增强版双向通信（带协议）

自定义协议定义

// 协议格式: [消息类型]:[内容]
// 消息类型: MSG, CMD, ERR, ACK
class MessageProtocol {
    static encode(type, content) {
        return `${type}:${content}`;
    }
    static decode(data) {
        const parts = data.toString().split(':');
        return {
            type: parts[0],
            content: parts.slice(1).join(':')
        };
    }
}

增强版服务器

const net = require('net');
const MessageProtocol = require('./protocol');
class TCPServer {
    constructor(port) {
        this.port = port;
        this.clients = new Map();
        this.server = null;
    }
    start() {
        this.server = net.createServer((socket) => {
            const clientId = `${socket.remoteAddress}:${socket.remotePort}`;
            console.log(`新客户端连接: ${clientId}`);
            // 存储客户端
            this.clients.set(clientId, {
                socket,
                address: socket.remoteAddress,
                port: socket.remotePort,
                connectedAt: new Date()
            });
            // 发送欢迎消息
            const welcome = MessageProtocol.encode('MSG', `欢迎 ${clientId}!`);
            socket.write(welcome);
            // 处理数据
            socket.on('data', (data) => {
                try {
                    const message = MessageProtocol.decode(data);
                    this.handleMessage(clientId, message);
                } catch (err) {
                    const errorMsg = MessageProtocol.encode('ERR', '消息格式错误');
                    socket.write(errorMsg);
                }
            });
            // 处理断开
            socket.on('end', () => {
                console.log(`客户端断开: ${clientId}`);
                this.clients.delete(clientId);
                this.broadcast(`用户 ${clientId} 已离线`);
            });
            // 处理错误
            socket.on('error', (err) => {
                console.error(`客户端错误 ${clientId}:`, err.message);
                this.clients.delete(clientId);
            });
        });
        this.server.listen(this.port, () => {
            console.log(`TCP服务器启动在端口 ${this.port}`);
        });
    }
    handleMessage(clientId, message) {
        switch(message.type) {
            case 'MSG':
                console.log(`[${clientId}]: ${message.content}`);
                // 广播消息给所有客户端
                this.broadcast(`[${clientId}]: ${message.content}`, clientId);
                break;
            case 'CMD':
                this.handleCommand(clientId, message.content);
                break;
            case 'ACK':
                console.log(`确认消息: ${message.content}`);
                break;
            default:
                const errorMsg = MessageProtocol.encode('ERR', '未知消息类型');
                const client = this.clients.get(clientId);
                if (client) client.socket.write(errorMsg);
        }
    }
    handleCommand(clientId, command) {
        const client = this.clients.get(clientId);
        if (!client) return;
        switch(command.trim().toUpperCase()) {
            case 'HELP':
                const helpMsg = MessageProtocol.encode('MSG', '可用命令: HELP, TIME, LIST, EXIT');
                client.socket.write(helpMsg);
                break;
            case 'TIME':
                const timeMsg = MessageProtocol.encode('MSG', `服务器时间: ${new Date().toLocaleString()}`);
                client.socket.write(timeMsg);
                break;
            case 'LIST':
                const userList = Array.from(this.clients.keys()).join(', ');
                const listMsg = MessageProtocol.encode('MSG', `在线用户: ${userList}`);
                client.socket.write(listMsg);
                break;
            case 'EXIT':
                client.socket.end();
                break;
            default:
                const errorMsg = MessageProtocol.encode('ERR', `未知命令: ${command}`);
                client.socket.write(errorMsg);
        }
    }
    broadcast(message, excludeClientId = null) {
        this.clients.forEach((client, id) => {
            if (id !== excludeClientId) {
                const msg = MessageProtocol.encode('MSG', message);
                client.socket.write(msg);
            }
        });
    }
    stop() {
        this.clients.forEach((client) => {
            client.socket.end();
        });
        this.server.close();
        console.log('服务器已停止');
    }
}
// 启动服务器
const server = new TCPServer(8080);
server.start();

增强版客户端

const net = require('net');
const readline = require('readline');
const MessageProtocol = require('./protocol');
class TCPClient {
    constructor(host, port) {
        this.host = host;
        this.port = port;
        this.socket = null;
        this.username = '';
        this.connected = false;
    }
    connect() {
        this.socket = new net.Socket();
        this.socket.connect(this.port, this.host, () => {
            this.connected = true;
            console.log(`已连接到 ${this.host}:${this.port}`);
            this.startUserInput();
        });
        this.socket.on('data', (data) => {
            this.handleMessage(data);
        });
        this.socket.on('close', () => {
            this.connected = false;
            console.log('连接已关闭');
            process.exit(0);
        });
        this.socket.on('error', (err) => {
            console.error('连接错误:', err.message);
            this.connected = false;
        });
    }
    handleMessage(data) {
        try {
            const message = MessageProtocol.decode(data);
            switch(message.type) {
                case 'MSG':
                    console.log(`\n${message.content}`);
                    break;
                case 'CMD':
                    this.executeCommand(message.content);
                    break;
                case 'ERR':
                    console.error(`\n错误: ${message.content}`);
                    break;
                case 'ACK':
                    console.log(`\n确认: ${message.content}`);
                    break;
                default:
                    console.log(`\n未知消息: ${data.toString()}`);
            }
            // 显示输入提示
            if (this.connected) {
                process.stdout.write('\n输入消息: ');
            }
        } catch (err) {
            console.error('解析消息失败:', err.message);
        }
    }
    startUserInput() {
        const rl = readline.createInterface({
            input: process.stdin,
            output: process.stdout
        });
        // 获取用户名
        this.getUsername(rl);
    }
    getUsername(rl) {
        rl.question('请输入用户名: ', (username) => {
            this.username = username.trim();
            console.log(`你好, ${this.username}!`);
            console.log('输入 /help 查看可用命令');
            console.log('输入 /exit 退出');
            process.stdout.write('输入消息: ');
            rl.on('line', (input) => {
                if (!input.trim()) return;
                if (input.startsWith('/')) {
                    // 命令处理
                    const command = input.slice(1);
                    if (command.toLowerCase() === 'exit') {
                        console.log('断开连接...');
                        this.socket.end();
                        rl.close();
                        return;
                    }
                    // 发送命令到服务器
                    const cmdMsg = MessageProtocol.encode('CMD', command);
                    this.socket.write(cmdMsg);
                } else {
                    // 普通消息
                    const msg = MessageProtocol.encode('MSG', input);
                    this.socket.write(msg);
                }
                process.stdout.write('输入消息: ');
            });
        });
    }
    executeCommand(command) {
        switch(command.toUpperCase()) {
            case 'CLEAR':
                console.clear();
                break;
            default:
                console.log(`未知客户端命令: ${command}`);
        }
    }
    disconnect() {
        if (this.socket) {
            this.socket.end();
        }
    }
}
// 使用客户端
const client = new TCPClient('localhost', 8080);
client.connect();
// 优雅退出
process.on('SIGINT', () => {
    console.log('\n正在退出...');
    client.disconnect();
    process.exit();
});

Python版双向通信

Python服务器

import socket
import threading
from datetime import datetime
class TCPServer:
    def __init__(self, host='localhost', port=8080):
        self.host = host
        self.port = port
        self.server_socket = None
        self.clients = {}
    def start(self):
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.server_socket.bind((self.host, self.port))
        self.server_socket.listen(5)
        print(f"服务器启动在 {self.host}:{self.port}")
        try:
            while True:
                client_socket, address = self.server_socket.accept()
                print(f"新客户端连接: {address}")
                # 为每个客户端创建线程
                client_thread = threading.Thread(
                    target=self.handle_client,
                    args=(client_socket, address)
                )
                client_thread.start()
        except KeyboardInterrupt:
            print("\n服务器关闭中...")
        finally:
            self.stop()
    def handle_client(self, client_socket, address):
        client_id = f"{address[0]}:{address[1]}"
        self.clients[client_id] = client_socket
        # 发送欢迎消息
        welcome = f"MSG:欢迎 {client_id} 连接到服务器!"
        client_socket.send(welcome.encode())
        while True:
            try:
                data = client_socket.recv(1024)
                if not data:
                    break
                message = data.decode()
                print(f"[{client_id}]: {message}")
                # 处理消息
                response = self.process_message(client_id, message)
                client_socket.send(response.encode())
            except ConnectionResetError:
                break
            except Exception as e:
                print(f"错误: {e}")
                break
        # 客户端断开
        print(f"客户端断开: {client_id}")
        del self.clients[client_id]
        client_socket.close()
    def process_message(self, client_id, message):
        if message.startswith("CMD:"):
            command = message[4:].strip().upper()
            return self.handle_command(client_id, command)
        elif message.startswith("MSG:"):
            content = message[4:]
            # 广播给其他客户端
            self.broadcast(f"MSG:[{client_id}]: {content}", client_id)
            return f"ACK:消息已发送"
        else:
            return f"MSG:收到: {message}"
    def handle_command(self, client_id, command):
        if command == "TIME":
            return f"MSG:服务器时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        elif command == "LIST":
            clients = ', '.join(self.clients.keys())
            return f"MSG:在线用户: {clients}"
        elif command == "HELP":
            return "MSG:可用命令: TIME, LIST, HELP"
        else:
            return f"MSG:未知命令: {command}"
    def broadcast(self, message, exclude_client=None):
        for client_id, client_socket in self.clients.items():
            if client_id != exclude_client:
                try:
                    client_socket.send(message.encode())
                except:
                    pass
    def stop(self):
        for client_socket in self.clients.values():
            client_socket.close()
        if self.server_socket:
            self.server_socket.close()
        print("服务器已停止")
# 启动服务器
if __name__ == "__main__":
    server = TCPServer()
    server.start()

Python客户端

import socket
import threading
import sys
class TCPClient:
    def __init__(self, host='localhost', port=8080):
        self.host = host
        self.port = port
        self.client_socket = None
        self.connected = False
    def connect(self):
        try:
            self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.client_socket.connect((self.host, self.port))
            self.connected = True
            print(f"已连接到 {self.host}:{self.port}")
            # 启动接收消息线程
            receive_thread = threading.Thread(target=self.receive_messages)
            receive_thread.daemon = True
            receive_thread.start()
            # 处理用户输入
            self.handle_input()
        except Exception as e:
            print(f"连接失败: {e}")
        finally:
            self.disconnect()
    def receive_messages(self):
        while self.connected:
            try:
                data = self.client_socket.recv(1024)
                if not data:
                    break
                message = data.decode()
                if message.startswith("MSG:"):
                    print(f"\n{message[4:]}")
                elif message.startswith("ACK:"):
                    print(f"\n确认: {message[4:]}")
                elif message.startswith("ERR:"):
                    print(f"\n错误: {message[4:]}")
                else:
                    print(f"\n服务器: {message}")
                print("\n输入消息: ", end='', flush=True)
            except ConnectionResetError:
                break
            except Exception as e:
                print(f"\n接收错误: {e}")
                break
        self.connected = False
    def handle_input(self):
        print("输入 /help 查看命令")
        print("输入 /exit 退出")
        while self.connected:
            try:
                user_input = input("\n输入消息: ")
                if not user_input.strip():
                    continue
                if user_input.startswith('/'):
                    command = user_input[1:].strip()
                    if command.lower() == 'exit':
                        break
                    else:
                        self.client_socket.send(f"CMD:{command}".encode())
                else:
                    self.client_socket.send(f"MSG:{user_input}".encode())
            except KeyboardInterrupt:
                break
            except Exception as e:
                print(f"输入错误: {e}")
                break
    def disconnect(self):
        self.connected = False
        if self.client_socket:
            self.client_socket.close()
        print("已断开连接")
# 启动客户端
if __name__ == "__main__":
    client = TCPClient()
    client.connect()

测试示例

多客户端测试脚本

// test_tcp.js
const net = require('net');
// 启动多个客户端测试
function createTestClient(name, port) {
    const client = new net.Socket();
    client.connect(port, 'localhost', () => {
        console.log(`${name} 已连接`);
        // 发送测试消息
        client.write(`MSG:你好，我是 ${name}`);
        // 延迟发送第二条消息
        setTimeout(() => {
            client.write(`MSG:${name} 的第二条消息`);
        }, 2000);
        // 延迟断开连接
        setTimeout(() => {
            client.end();
        }, 5000);
    });
    client.on('data', (data) => {
        console.log(`${name} 收到: ${data.toString()}`);
    });
    client.on('close', () => {
        console.log(`${name} 连接关闭`);
    });
    return client;
}
// 创建3个测试客户端
createTestClient('客户端A', 8080);
createTestClient('客户端B', 8080);
createTestClient('客户端C', 8080);

关键特性和最佳实践

1. 双向通信: 客户端和服务器都可以主动发送消息
2. 消息协议: 使用自定义协议区分不同类型的消息
3. 并发处理: 支持多客户端同时连接
4. 错误处理: 完善的异常处理和连接管理
5. 用户界面: 提供命令行交互界面

最佳实践

1. 消息缓冲区: 处理大消息时需要缓冲区
2. 心跳机制: 定期发送心跳包检测连接状态
3. 重连机制: 实现自动重连逻辑
4. 消息队列: 使用队列管理待发送消息
5. 流量控制: 实现背压机制防止数据溢出

这些示例提供了完整的TCP双向通信实现,你可以根据自己的需求进行修改和扩展。