本文目录导读：
我来详细讲解如何使用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);
关键特性和最佳实践
- 双向通信: 客户端和服务器都可以主动发送消息
- 消息协议: 使用自定义协议区分不同类型的消息
- 并发处理: 支持多客户端同时连接
- 错误处理: 完善的异常处理和连接管理
- 用户界面: 提供命令行交互界面
最佳实践
- 消息缓冲区: 处理大消息时需要缓冲区
- 心跳机制: 定期发送心跳包检测连接状态
- 重连机制: 实现自动重连逻辑
- 消息队列: 使用队列管理待发送消息
- 流量控制: 实现背压机制防止数据溢出
这些示例提供了完整的TCP双向通信实现,你可以根据自己的需求进行修改和扩展。
