Node.js fs单文件异步互斥读写优化方案咨询

在API场景中，多个HTTP请求会对单个文件进行读写操作，请求同步触发。为避免文件访问冲突，实现了一个异步PromiseChain，但仍出现文件未操作完成就被访问的情况。希望通过全局持续文件流解决冲突并提升性能，让请求处理器以链式异步方式接入流中，熟悉RxJS但对fs流不熟悉，需要入门指引。

一、PromiseChain失效的可能原因 #

你的PromiseChain逻辑本身是串行化的，但问题可能出在：

• 路由处理器未将完整的原子性文件操作推入链中（比如readFile和writeFile被拆分为独立操作，中间存在插队风险）
• 函数调用存在参数不匹配问题（比如oneFileOperation定义需要data参数，但调用时未传递）

二、RxJS + 内存状态流方案（适合JSON/结构化小文件） #

核心思路是维护一个全局内存状态流，所有读写操作串行处理，仅在必要时写回文件，既避免冲突又提升性能。

1. 完整实现代码（TypeScript） #

import { BehaviorSubject, concatMap, tap, debounceTime } from 'rxjs';
import { readFile, writeFile } from 'fs/promises';

// 定义文件数据类型
type FileData = Record<string, any>;

// 全局内存状态：保存当前文件的最新数据
const fileState$ = new BehaviorSubject<FileData>({});
// 全局操作队列：用concatMap保证所有操作串行执行
const operationQueue$ = new BehaviorSubject<(state: FileData) => Promise<FileData>>(async () => fileState$.value);

// 初始化：服务启动时读取文件到内存
(async () => {
  try {
    const content = await readFile('my-file.json', 'utf-8');
    fileState$.next(JSON.parse(content));
  } catch (e) {
    // 文件不存在时初始化空对象
    if ((e as NodeJS.ErrnoException).code === 'ENOENT') {
      fileState$.next({});
      await writeFile('my-file.json', JSON.stringify({}));
    } else {
      throw e;
    }
  }
})();

// 订阅操作队列，串行执行并同步文件
operationQueue$
  .pipe(
    concatMap(async (operation) => {
      const currentState = fileState$.value;
      // 执行操作，返回新状态
      return await operation(currentState);
    }),
    // 更新内存状态
    tap((newState) => fileState$.next(newState)),
    // 防抖：避免短时间内多次写文件（可根据需求调整时长）
    debounceTime(100),
    // 将最新状态写回文件
    tap(async (newState) => {
      await writeFile('my-file.json', JSON.stringify(newState, null, 2));
    })
  )
  .subscribe({
    error: (err) => console.error('文件操作出错:', err),
  });

// 暴露给路由的操作入口
export function addFileOperation(operation: (state: FileData) => Promise<FileData>) {
  operationQueue$.next(operation);
}

2. 路由中使用示例 #

// 读取操作：获取文件全部数据
async function getFileData() {
  return new Promise<FileData>((resolve) => {
    addFileOperation(async (state) => {
      resolve(state);
      return state; // 不修改状态，直接返回
    });
  });
}

// 写入操作：修改指定字段
async function updateFileData(key: string, value: any) {
  return new Promise<void>((resolve) => {
    addFileOperation(async (state) => {
      state[key] = value;
      resolve();
      return state;
    });
  });
}

// 路由处理器示例
// GET /api/data
async function getDataRoute() {
  const data = await getFileData();
  return { status: 200, body: data };
}

// POST /api/data
async function updateDataRoute(req: any) {
  await updateFileData(req.body.key, req.body.value);
  return { status: 200, body: { message: '更新成功' } };
}

三、大文件场景：RxJS + FS流方案 #

如果处理超大文件（如日志、非结构化文本），需要用FS流结合RxJS串行处理：

import { BehaviorSubject, concatMap, from } from 'rxjs';
import { createReadStream, createWriteStream } from 'fs';
import { pipeline } from 'stream/promises';

// 全局流操作队列
const streamOperationQueue$ = new BehaviorSubject<() => Promise<void>>(async () => {});

// 串行执行所有流操作
streamOperationQueue$
  .pipe(concatMap((op) => from(op())))
  .subscribe({ error: (err) => console.error('流操作出错:', err) });

// 读取流操作示例
function addReadStreamTask() {
  streamOperationQueue$.next(async () => {
    const readStream = createReadStream('large-file.txt');
    // 逐块处理流数据
    for await (const chunk of readStream) {
      console.log(chunk.toString());
    }
  });
}

// 写入流操作示例（追加模式）
function addWriteStreamTask(data: string) {
  streamOperationQueue$.next(async () => {
    const writeStream = createWriteStream('large-file.txt', { flags: 'a' });
    await pipeline(from([data]), writeStream);
  });
}

注：处理超大JSON需搭配JSONStream库解析，避免内存溢出。

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