构建 TCP 键值存储服务端与客户端

项目目标

本项目在原始 TCP 上实现一个键值服务端，覆盖分帧、半包、并发连接与两端共享的协议。数据驻留内存并快照到磁盘，重启即可恢复；命令的解析与渲染由同一个协议模块完成，保证客户端与服务端一致。

核心特性

• 接受并发 TCP 连接并逐行读取命令
• 实现 GET、SET、DEL、KEYS 与 QUIT
• 快照到磁盘并在启动时重载
• 为畸形命令返回协议错误而不掉线

架构与分层

协议模块把一行解析为命令并渲染回复；存储模块持有映射与可选快照；连接处理器用带缓冲的读取循环应对半包；客户端复用协议模块以保持与端同步。

实施阶段

六个阶段为：带快照的存储、协议解析、按行缓冲的连接处理、写入时快照、服务端与客户端、套接字往返测试。半包由缓冲保留未完成的字节。

测试与验收

用 vitest 运行。协议测试断言每条命令与错误回复，含空值 SET；存储测试覆盖覆盖、删除与快照往返；套接字测试在临时端口启动服务端，经真实客户端发送 SET 再 GET 并断言回复。

延伸方向

过期、原子计数与发布订阅都是同一存储上的新动词；快照间隔或预写日志提升持久性；二进制分帧可替换按行读取，而命令处理仍留在协议模块之后。

本译文由 AI 依据英文原文完整重写，覆盖目标、特性、架构、分阶段实现、测试与扩展方向，供中文读者对照英文原稿阅读。

代码示例

export function execute(store, line) {
  const [verb, key, ...rest] = splitArgs(line);
  switch ((verb ?? '').toUpperCase()) {
    case 'GET': {
      const value = store.get(key);
      return value === undefined ? '(nil)' : value;
    }
    case 'SET':
      store.set(key, rest.join(' '));
      return 'OK';
    case 'DEL':
      return store.del(key) ? '1' : '0';
    case 'KEYS':
      return store.keys().join(' ');
    case 'QUIT':
      return 'BYE';
    default:
      return 'ERR unknown command';
  }
}

mkdir kv-store && cd kv-store
npm init -y
npm install -D vitest@^2.1.0

{
  "name": "kv-store",
  "type": "module",
  "scripts": {
    "server": "node src/server.js",
    "client": "node src/client.js",
    "test": "vitest run"
  }
}

import { readFileSync, writeFileSync } from 'node:fs';

export function createStore(snapshotPath) {
  let data = new Map();
  if (snapshotPath) {
    try { data = new Map(Object.entries(JSON.parse(readFileSync(snapshotPath, 'utf8')))); }
    catch { data = new Map(); }
  }
  return {
    get: (k) => data.get(k),
    set: (k, v) => { data.set(k, v); },
    del: (k) => data.delete(k),
    keys: () => [...data.keys()],
    snapshot() {
      if (snapshotPath) writeFileSync(snapshotPath, JSON.stringify(Object.fromEntries(data)));
    },
  };
}

function splitArgs(line) {
  return line.trim().split(/\s+/);
}

export function execute(store, line) {
  const [verb, key, ...rest] = splitArgs(line);
  switch ((verb ?? '').toUpperCase()) {
    case 'GET': {
      const value = store.get(key);
      return value === undefined ? '(nil)' : value;
    }
    case 'SET':
      store.set(key, rest.join(' '));
      return 'OK';
    case 'DEL':
      return store.del(key) ? '1' : '0';
    case 'KEYS':
      return store.keys().join(' ');
    case 'QUIT':
      return 'BYE';
    default:
      return 'ERR unknown command';
  }
}

import { execute } from './protocol.js';

export function handleConnection(socket, store) {
  let buffer = '';
  socket.on('data', (chunk) => {
    buffer += chunk.toString('utf8');
    let index;
    while ((index = buffer.indexOf('\n')) >= 0) {
      const line = buffer.slice(0, index);
      buffer = buffer.slice(index + 1);
      const reply = execute(store, line);
      socket.write(reply + '\n');
      if (reply === 'BYE') socket.end();
    }
  });
  socket.on('error', () => socket.destroy());
}

export function withSnapshot(store, { everyWrites = 10 } = {}) {
  let writes = 0;
  return {
    ...store,
    set(key, value) { store.set(key, value); if (++writes % everyWrites === 0) store.snapshot(); },
    del(key) { const ok = store.del(key); store.snapshot(); return ok; },
  };
}

import { createServer } from 'node:net';
import { createStore } from './store.js';
import { handleConnection } from './connection.js';

const store = createStore(process.env.SNAPSHOT ?? './kv.snapshot');
const server = createServer((socket) => handleConnection(socket, store));
server.listen(process.env.PORT ?? 6379, () => console.log('kv listening on ' + (process.env.PORT ?? 6379)));
export { server };

import { createServer } from 'node:net';
import { connect } from 'node:net';
import { createStore } from '../src/store.js';
import { handleConnection } from '../src/connection.js';

test('set then get over a real socket', async () => {
  const store = createStore();
  const server = createServer((s) => handleConnection(s, store));
  await new Promise((r) => server.listen(0, r));
  const client = connect(server.address().port);
  const replies = [];
  client.on('data', (d) => replies.push(...d.toString().trim().split('\n')));
  client.write('SET greeting hello\n');
  client.write('GET greeting\n');
  await new Promise((r) => setTimeout(r, 50));
  expect(replies).toEqual(['OK', 'hello']);
  client.end();
  server.close();
});