怎样使用Node.js操作共享内存？

worker_threads

SharedArrayBuffer

SharedArrayBuffer

worker_threads

Atomics

shm_open

mmap

CreateFileMapping

MapViewOfFile

worker_threads

SharedArrayBuffer

Atomics

SharedArrayBuffer

worker_threads

ArrayBuffer

SharedArrayBuffer

// main.js (主线程)
const { Worker, isMainThread, parentPort } = require('worker_threads');

if (isMainThread) {
  // 创建一个1KB的共享内存缓冲区
  const sharedBuffer = new SharedArrayBuffer(1024);
  const sharedArray = new Int32Array(sharedBuffer);

  // 初始化共享内存的第一个元素
  sharedArray[0] = 0;
  console.log('主线程：初始值 =', sharedArray[0]);

  // 启动两个Worker
  const worker1 = new Worker(__filename);
  const worker2 = new Worker(__filename);

  // 将共享缓冲区传递给Worker
  worker1.postMessage({ sharedBuffer });
  worker2.postMessage({ sharedBuffer });

  let completedWorkers = 0;
  worker1.on('message', (msg) => {
    if (msg === 'done') {
      completedWorkers++;
      if (completedWorkers === 2) {
        console.log('主线程：所有Worker完成，最终值 =', sharedArray[0]);
      }
    }
  });
  worker2.on('message', (msg) => {
    if (msg === 'done') {
      completedWorkers++;
      if (completedWorkers === 2) {
        console.log('主线程：所有Worker完成，最终值 =', sharedArray[0]);
      }
    }
  });

} else {
  // worker.js (工作线程)
  parentPort.on('message', (msg) => {
    const { sharedBuffer } = msg;
    const sharedArray = new Int32Array(sharedBuffer);

    // 每个Worker对共享内存的第一个元素进行1000次原子加操作
    for (let i = 0; i < 1000; i++) {
      Atomics.add(sharedArray, 0, 1); // 原子加操作
    }
    parentPort.postMessage('done');
  });
}

sharedArray[0]

shared_mem_addon.cc

#include <napi.h>
#include <iostream>
#include <string>
#include <vector>

#ifdef _WIN32
#include <windows.h>
#else
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#endif

// 这是一个非常简化的示例，实际生产环境需要更健壮的错误处理和资源管理
void* shared_memory_ptr = nullptr;
size_t shared_memory_size = 0;
#ifdef _WIN32
HANDLE hMapFile = NULL;
#else
int shm_fd = -1;
#endif

Napi::Value CreateSharedMemory(const Napi::CallbackInfo& info) {
    Napi::Env env = info.Env();
    if (info.Length() < 2 || !info[0].IsString() || !info[1].IsNumber()) {
        Napi::TypeError::New(env, "Expected shared memory key (string) and size (number)").ThrowAsJavaScriptException();
        return env.Null();
    }

    std::string key = info[0].As<Napi::String>().Utf8Value();
    shared_memory_size = info[1].As<Napi::Number>().Uint32Value();

#ifdef _WIN32
    hMapFile = CreateFileMapping(
        INVALID_HANDLE_VALUE,    // use paging file
        NULL,                    // default security
        PAGE_READWRITE,          // read/write access
        0,                       // maximum object size (high-order DWORD)
        shared_memory_size,      // maximum object size (low-order DWORD)
        key.c_str());            // name of mapping object

    if (hMapFile == NULL) {
        Napi::Error::New(env, "Could not create file mapping object").ThrowAsJavaScriptException();
        return env.Null();
    }

    shared_memory_ptr = MapViewOfFile(
        hMapFile,                // handle to map object
        FILE_MAP_ALL_ACCESS,     // read/write access
        0,
        0,
        shared_memory_size);

    if (shared_memory_ptr == NULL) {
        CloseHandle(hMapFile);
        Napi::Error::New(env, "Could not map view of file").ThrowAsJavaScriptException();
        return env.Null();
    }
#else // POSIX
    shm_fd = shm_open(key.c_str(), O_CREAT | O_RDWR, 0666);
    if (shm_fd == -1) {
        Napi::Error::New(env, "shm_open failed").ThrowAsJavaScriptException();
        return env.Null();
    }

    if (ftruncate(shm_fd, shared_memory_size) == -1) {
        close(shm_fd);
        Napi::Error::New(env, "ftruncate failed").ThrowAsJavaScriptException();
        return env.Null();
    }

    shared_memory_ptr = mmap(0, shared_memory_size, PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
    if (shared_memory_ptr == MAP_FAILED) {
        close(shm_fd);
        Napi::Error::New(env, "mmap failed").ThrowAsJavaScriptException();
        return env.Null();
    }
#endif

    return Napi::Boolean::New(env, true);
}

Napi::Value GetSharedMemoryBuffer(const Napi::CallbackInfo& info) {
    Napi::Env env = info.Env();
    if (shared_memory_ptr == nullptr) {
        Napi::Error::New(env, "Shared memory not initialized").ThrowAsJavaScriptException();
        return env.Null();
    }
    // N-API允许从原生内存创建ArrayBuffer
    return Napi::ArrayBuffer::New(env, shared_memory_ptr, shared_memory_size);
}

// 在模块卸载时清理资源
Napi::Object Init(Napi::Env env, Napi::Object exports) {
    exports.Set(Napi::String::New(env, "createSharedMemory"), Napi::Function::New(env, CreateSharedMemory));
    exports.Set(Napi::String::New(env, "getSharedMemoryBuffer"), Napi::Function::New(env, GetSharedMemoryBuffer));
    return exports;
}

NODE_API_MODULE(shared_mem_addon, Init)

app.js

// app.js
const sharedMemAddon = require('./build/Release/shared_mem_addon'); // 假设已编译好

const SHM_KEY = 'my_unique_shm_key';
const SHM_SIZE = 1024; // 1KB

try {
  // 创建或连接共享内存
  const created = sharedMemAddon.createSharedMemory(SHM_KEY, SHM_SIZE);
  console.log(`共享内存 ${SHM_KEY} ${created ? '创建/连接成功' : '失败'}`);

  // 获取共享内存的ArrayBuffer视图
  const sharedBuffer = sharedMemAddon.getSharedMemoryBuffer();
  const uint8Array = new Uint8Array(sharedBuffer);

  // 写入数据
  const message = "Hello from Node.js!";
  for (let i = 0; i < message.length; i++) {
    uint8Array[i] = message.charCodeAt(i);
  }
  uint8Array[message.length] = 0; // null terminator

  console.log('Node.js写入数据:', message);

  // 假设有另一个进程（可能是另一个Node.js进程或C++程序）
  // 也能通过相同的SHM_KEY连接到这块共享内存并读取数据。
  // ...
  // 在真实场景中，这里还需要一个机制来同步读写，比如信号量或互斥锁，
  // 同样可以在C++ addon中实现并暴露给JS。

} catch (error) {
  console.error('操作共享内存失败:', error.message);
}

// 注意：共享内存的清理（如shm_unlink或UnmapViewOfFile/CloseHandle）
// 同样需要通过N-API在C++中实现并由Node.js调用，
// 或者在进程退出时自动清理（依赖OS行为）。