Node의 프로세스 간 통신에 대해 이야기하는 기사

프로세스 간 통신은 어떻게 하나요? 다음 글에서는 Node 프로세스간 통신의 원리를 소개하겠습니다. 도움이 되셨으면 좋겠습니다!

전제 지식

파일 설명자

리눅스 시스템에서는 모든 것이 파일로 취급되며 프로세스가 기존 파일을 열면 파일 설명자가 반환됩니다. 파일 설명자는 프로세스에 의해 열린 파일을 관리하기 위해 운영 체제에서 생성된 인덱스이며 열린 파일을 가리키는 데 사용됩니다. 프로세스가 시작되면 운영 체제는 각 프로세스에 PCB 제어 블록을 할당합니다. PCB에는 현재 프로세스의 모든 파일 설명자, 즉 현재 프로세스에서 열린 모든 파일을 저장하는 파일 설명자 테이블이 있습니다.

프로세스의 파일 설명자는 시스템 파일과 어떻게 일치합니까? 커널에서 시스템은 두 개의 다른 테이블을 유지 관리합니다.

• Open 파일 테이블
• i-node 테이블(i-node 테이블)

파일 설명자는 0부터 시작하여 0/씩 증가하는 배열의 첨자입니다. 1/2는 기본적으로 입력/출력/오류 스트림의 파일 설명자입니다. PCB에 유지되는 파일 설명 테이블에서는 파일 설명자와 해당 오픈 파일 테이블을 기반으로 해당 파일 포인터를 찾을 수 있습니다. 열린 파일 테이블은 다음을 유지합니다: 파일 오프셋(파일을 읽고 쓸 때 업데이트됨), i-노드 테이블에 대한 파일 포인터에 대한 상태 식별자 실제로 파일을 조작하려면 i-node 테이블에 의존하여 실제 파일에 대한 관련 정보를 얻어야 합니다

그들 사이의 관계

Illustration

• 프로세스 A에서는 파일 디스크립터 1/20은 모두 동일한 열린 파일 테이블 항목 23을 가리킵니다. 이는 동일한 파일에서 열기 함수를 여러 번 호출하여 발생할 수 있습니다. 프로세스 A/B의 파일 설명자 2가 모두 동일한 파일을 가리킬 수 있습니다. 프로세스 A의 파일 설명자 0과 프로세스 B의 파일 설명자는 서로 다른 열린 파일 테이블 항목을 가리키지만 이러한 테이블 항목은 다음을 가리킵니다. 동일한 파일, 이는 A/B일 수 있습니다. 프로세스가 각각 동일한 파일에 대한 공개 호출을 시작했습니다
• 요약

동일 프로세스의 서로 다른 파일 설명자가 동일한 파일을 가리킬 수 있습니다.

다양한 프로세스가 동일한 파일을 가질 수 있습니다. 파일 설명자

• 다른 프로세스의 동일한 파일 설명자가 다른 파일을 가리킬 수 있음
• 다른 프로세스의 서로 다른 파일 설명자가 동일한 파일을 가리킬 수 있음
• 파일 설명자의 리디렉션

프로세스를 읽거나 쓸 때마다 파일 설명자로 시작하여 해당 항목을 찾습니다. 파일 테이블 항목을 열고 해당 i-node 테이블을 찾습니다. 파일 설명자 리디렉션을 구현하는 방법은 무엇입니까? 해당 파일 포인터는 파일 설명자 테이블에서 찾을 수 있기 때문입니다. 파일 포인터를 변경하면 다음 두 테이블의 내용도 변경됩니까? 예를 들어, 파일 설명자 1은 모니터를 가리키고, 파일 설명자 1은 log.txt 파일을 가리킨 다음, 파일 설명자 1은 log.txt

파일 설명자의 쉘 리디렉션

에 해당합니다. > , 및

c 함수는 파일 설명자를 리디렉션합니다

cat hello.txt时，默认会将结果输出到显示器上，使用 > 来重定向。cat hello.txt 1 > log.txt

dup

dup 함수는 oldfd와 동일한 파일을 가리키며 파일 오프셋 및 파일 상태를 공유하는 데 사용됩니다

int main(int argc, char const *argv[])
{
    int fd = open("log.txt");
    int copyFd = dup(fd);
    //将fd阅读文件置于文件末尾，计算偏移量。
    cout <strong></strong> 

dup(3)을 호출하면 새로운 최소 설명자 4가 열립니다. 이 4는 3이 가리키는 파일을 가리킵니다. fd에 대한 모든 작업은 파일을 수정합니다

dup2

dup2 기능 포인트 oldfd가 가리키는 파일에 지정된 newfd를 추가합니다. dup2 실행 후 newfd와 oldfd가 동시에 동일한 파일을 가리키며 파일 오프셋 및 파일 상태를 공유합니다

int main(int argc, char const *argv[])
{
    int fd = open("log.txt");
    int copyFd = dup(fd);
    //将fd阅读文件置于文件末尾，计算偏移量。
    cout <p></p><h2 id="strong-Node中通信原理-strong"><strong>Node中通信原理</strong></h2><p>Node 中的 IPC 通道具体实现是由 <a href="https://link.juejin.cn?target=https%3A%2F%2Fluohaha.github.io%2FChinese-uvbook%2Fsource%2Fintroduction.html" target="_blank" rel="nofollow noopener noreferrer" title="https://luohaha.github.io/Chinese-uvbook/source/introduction.html" ref="nofollow noopener noreferrer">libuv</a> 提供的。根据系统的不同实现方式不同，window 下采用命名管道实现，*nix 下采用 Domain Socket 实现。在应用层只体现为 message 事件和 send 方法。【相关教程推荐：<a href="https://www.php.cn/course/list/24.html" target="_blank">nodejs视频教程</a>】</p><p><img src="/static/imghw/default1.png" data-src="https://img.php.cn/upload/image/370/681/147/1662375066320124.png" class="lazy" title="166237505660155Node의 프로세스 간 통신에 대해 이야기하는 기사" alt="Node의 프로세스 간 통신에 대해 이야기하는 기사"></p><p>父进程在实际创建子进程之前，会创建 IPC 通道并监听它，等到创建出真实的子进程后，通过环境变量(NODE_CHANNEL_FD)告诉子进程该 IPC 通道的文件描述符。</p><p>子进程在启动的过程中，会根据该文件描述符去连接 IPC 通道，从而完成父子进程的连接。</p><p>建立连接之后可以自由的通信了，IPC 通道是使用命名管道或者 Domain Socket 创建的，属于双向通信。并且它是在系统内核中完成的进程通信</p><p><img src="/static/imghw/default1.png" data-src="https://img.php.cn/upload/image/370/681/147/1662375066320124.png" class="lazy" title="166237506175944Node의 프로세스 간 통신에 대해 이야기하는 기사" alt="Node의 프로세스 간 통신에 대해 이야기하는 기사"></p><p>⚠️ 只有在启动的子进程是 Node 进程时，子进程才会根据环境变量去连接对应的 IPC 通道，对于其他类型的子进程则无法实现进程间通信，除非其他进程也按着该约定去连接这个 IPC 通道。</p><h3 id="strong-unix-domain-socket-strong"><strong>unix domain socket</strong></h3><h4 id="strong-是什么-strong"><strong>是什么</strong></h4><p>我们知道经典的通信方式是有 Socket，我们平时熟知的 Socket 是基于网络协议的，用于两个不同主机上的两个进程通信，通信需要指定 IP/Host 等。
但如果我们同一台主机上的两个进程想要通信，如果使用 Socket 需要指定 IP/Host，经过网络协议等，会显得过于繁琐。所以 Unix Domain Socket 诞生了。</p><p>UDS 的优势：</p>

• 绑定 socket 文件而不是绑定 IP/Host；不需要经过网络协议，而是数据的拷贝
• 也支持 SOCK_STREAM(流套接字)和 SOCK_DGRAM(数据包套接字)，但由于是在本机通过内核通信，不会丢包也不会出现发送包的次序和接收包的次序不一致的问题

如何实现

流程图

Server 端

int main(int argc, char *argv[])
{
    int server_fd ,ret, client_fd;
    struct sockaddr_un serv, client;
    socklen_t len = sizeof(client);
    char buf[1024] = {0};
    int recvlen;

    // 创建 socket
    server_fd = socket(AF_LOCAL, SOCK_STREAM, 0);

    // 初始化 server 信息
    serv.sun_family = AF_LOCAL;
    strcpy(serv.sun_path, "server.sock");

    // 绑定
    ret = bind(server_fd, (struct sockaddr *)&serv, sizeof(serv));

    //设置监听，设置能够同时和服务端连接的客户端数量
    ret = listen(server_fd, 36);

    //等待客户端连接
    client_fd = accept(server_fd, (struct sockaddr *)&client, &len);
    printf("=====client bind file:%s\n", client.sun_path);

    while (1) {
        recvlen = recv(client_fd, buf, sizeof(buf), 0);
        if (recvlen == -1) {
            perror("recv error");
            return -1;
        } else if (recvlen == 0) {
            printf("client disconnet...\n");
            close(client_fd);
            break;
        } else {
            printf("recv buf %s\n", buf);
            send(client_fd, buf, recvlen, 0);
        }
    }

    close(client_fd);
    close(server_fd);
    return 0;
}

Client 端

int main(int argc, char *argv[])
{
    int client_fd ,ret;
    struct sockaddr_un serv, client;
    socklen_t len = sizeof(client);
    char buf[1024] = {0};
    int recvlen;

    //创建socket
    client_fd = socket(AF_LOCAL, SOCK_STREAM, 0);

    //给客户端绑定一个套接字文件
    client.sun_family = AF_LOCAL;
    strcpy(client.sun_path, "client.sock");
    ret = bind(client_fd, (struct sockaddr *)&client, sizeof(client));

    //初始化server信息
    serv.sun_family = AF_LOCAL;
    strcpy(serv.sun_path, "server.sock");
    //连接
    connect(client_fd, (struct sockaddr *)&serv, sizeof(serv));

    while (1) {
        fgets(buf, sizeof(buf), stdin);
        send(client_fd, buf, strlen(buf)+1, 0);

        recv(client_fd, buf, sizeof(buf), 0);
        printf("recv buf %s\n", buf);
    }

    close(client_fd);
    return 0;
}

命名管道(Named Pipe)

是什么

命名管道是可以在同一台计算机的不同进程之间，或者跨越一个网络的不同计算机的不同进程之间的可靠的单向或者双向的数据通信。 创建命名管道的进程被称为管道服务端(Pipe Server)，连接到这个管道的进程称为管道客户端(Pipe Client)。

命名管道的命名规范：\server\pipe[\path]\name

• 其中 server 指定一个服务器的名字，本机适用 \. 表示，\192.10.10.1 表示网络上的服务器
• \pipe 是一个不可变化的字串，用于指定该文件属于 NPFS(Named Pipe File System)
• [\path]\name 是唯一命名管道名称的标识

怎么实现

流程图

Pipe Server

void ServerTest()
{
    HANDLE  serverNamePipe;
    char    pipeName[MAX_PATH] = {0};
    char    szReadBuf[MAX_BUFFER] = {0};
    char    szWriteBuf[MAX_BUFFER] = {0};
    DWORD   dwNumRead = 0;
    DWORD   dwNumWrite = 0;

    strcpy(pipeName, "\\\\.\\pipe\\shuangxuPipeTest");
    // 创建管道实例
    serverNamePipe = CreateNamedPipeA(pipeName,
        PIPE_ACCESS_DUPLEX|FILE_FLAG_WRITE_THROUGH,
        PIPE_TYPE_BYTE|PIPE_READMODE_BYTE|PIPE_WAIT,
        PIPE_UNLIMITED_INSTANCES, 0, 0, 0, NULL);
    WriteLog("创建管道成功...");
    // 等待客户端连接
    BOOL bRt= ConnectNamedPipe(serverNamePipe, NULL );
    WriteLog( "收到客户端的连接成功...");
    // 接收数据
    memset( szReadBuf, 0, MAX_BUFFER );
    bRt = ReadFile(serverNamePipe, szReadBuf, MAX_BUFFER-1, &dwNumRead, NULL );
    // 业务逻辑处理 （只为测试用返回原来的数据）
    WriteLog( "收到客户数据:[%s]", szReadBuf);
    // 发送数据
    if( !WriteFile(serverNamePipe, szWriteBuf, dwNumRead, &dwNumWrite, NULL ) )
    {
        WriteLog("向客户写入数据失败:[%#x]", GetLastError());
        return ;
    }
    WriteLog("写入数据成功...");
}

Pipe Client

void ClientTest()
{
    char    pipeName[MAX_PATH] = {0};
    HANDLE  clientNamePipe;
    DWORD   dwRet;
    char    szReadBuf[MAX_BUFFER] = {0};
    char    szWriteBuf[MAX_BUFFER] = {0};
    DWORD   dwNumRead = 0;
    DWORD   dwNumWrite = 0;

    strcpy(pipeName, "\\\\.\\pipe\\shuangxuPipeTest");
    // 检测管道是否可用
    if(!WaitNamedPipeA(pipeName, 10000)){
        WriteLog("管道[%s]无法打开", pipeName);
        return ;
    }
    // 连接管道
    clientNamePipe = CreateFileA(pipeName,
        GENERIC_READ|GENERIC_WRITE,
        0,
        NULL,
        OPEN_EXISTING,
        FILE_ATTRIBUTE_NORMAL,
        NULL);
    WriteLog("管道连接成功...");
    scanf( "%s", szWritebuf );
    // 发送数据
    if( !WriteFile(clientNamePipe, szWriteBuf, strlen(szWriteBuf), &dwNumWrite, NULL)){
        WriteLog("发送数据失败,GetLastError=[%#x]", GetLastError());
        return ;
    }
    printf("发送数据成功:%s\n", szWritebuf );
    // 接收数据
    if( !ReadFile(clientNamePipe, szReadBuf, MAX_BUFFER-1, &dwNumRead, NULL)){
        WriteLog("接收数据失败,GetLastError=[%#x]", GetLastError() );
        return ;
    }
    WriteLog( "接收到服务器返回:%s", szReadBuf );
    // 关闭管道
    CloseHandle(clientNamePipe);
}

Node 创建子进程的流程

Unix

对于创建子进程、创建管道、重定向管道均是在 c++ 层实现的

创建子进程

int main(int argc,char *argv[]){
    pid_t pid = fork();
    if (pid <h4 id="strong-创建管道-strong"><strong>创建管道</strong></h4><p>使用 socketpair 创建管道，其创建出来的管道是全双工的，返回的文件描述符中的任何一个都可读和可写</p><pre class="brush:php;toolbar:false">int main ()
{
    int fd[2];
    int r = socketpair(AF_UNIX, SOCK_STREAM, 0, fd);

    if (fork()){ /* 父进程 */
        int val = 0;
        close(fd[1]);
        while (1){
            sleep(1);
            ++val;
            printf("发送数据: %d\n", val);
            write(fd[0], &val, sizeof(val));
            read(fd[0], &val, sizeof(val));
            printf("接收数据: %d\n", val);
        }
    } else {  /*子进程*/
        int val;
        close(fd[0]);
        while(1){
            read(fd[1], &val, sizeof(val));
            ++val;
            write(fd[1], &val, sizeof(val));
        }
    }
}

当我们使用 socketpair 创建了管道之后，父进程关闭了 fd[1]，子进程关闭了 fd[0]。子进程可以通过 fd[1] 读写数据；同理主进程通过 fd[0]读写数据完成通信。

对应代码：https://github.com/nodejs/node/blob/main/deps/uv/src/unix/process.c#L344

child_process.fork 的详细调用

fork 函数开启一个子进程的流程

• 初始化参数中的 options.stdio，并且调用 spawn 函数

  function spawn(file, args, options) {
    const child = new ChildProcess();
  
    child.spawn(options);
  }

• 创建 ChildProcess 实例，创建子进程也是调用 C++ 层 this._handle.spawn 方法

  function ChildProcess() {
  	// C++层定义
  	this._handle = new Process();
  }

• 通过 child.spawn 调用到 ChildProcess.prototype.spawn 方法中。其中 getValidStdio 方法会根据 options.stdio 创建和 C++ 交互的 Pipe 对象，并获得对应的文件描述符，将文件描述符写入到环境变量 NODE_CHANNEL_FD 中，调用 C++ 层创建子进程，在调用 setupChannel 方法

  ChildProcess.prototype.spawn = function(options) {
    // 预处理进程间通信的数据结构
  	stdio = getValidStdio(stdio, false);
  	const ipc = stdio.ipc;
      const ipcFd = stdio.ipcFd;
  	//将文件描述符写入环境变量中
  	if (ipc !== undefined) {
      ArrayPrototypePush(options.envPairs, `NODE_CHANNEL_FD=${ipcFd}`);
    }
  	// 创建进程
  	const err = this._handle.spawn(options);
      // 添加send方法和监听IPC中数据
  	if (ipc !== undefined) setupChannel(this, ipc, serialization);
  }

• 子进程启动时，会根据环境变量中是否存在 NODE_CHANNEL_FD 判断是否调用 _forkChild 方法，创建一个 Pipe 对象, 同时调用 open 方法打开对应的文件描述符，在调用setupChannel

  function _forkChild(fd, serializationMode) {
    const p = new Pipe(PipeConstants.IPC);
    p.open(fd);
    p.unref();
    const control = setupChannel(process, p, serializationMode);
  }

句柄传递

setupChannel 主要是完成了处理接收的消息、发送消息、处理文件描述符传递等

function setipChannel(){
	channel.onread = function(arrayBuffer){
		//...
	}
	target.on('internalMessage', function(message, handle){
		//...
	})
	target.send = function(message, handle, options, callback){
		//...
	}
	target._send = function(message, handle, options, callback){
		//...
	}
	function handleMessage(message, handle, internal){
		//...
	}
}

• target.send: process.send 方法，这里 target 就是进程对象本身.
• target._send: 执行具体 send 逻辑的函数, 当参数 handle 不存在时, 表示普通的消息传递；若存在，包装为内部对象，表明是一个 internalMessage 事件触发。调用使用JSON.stringify 序列化对象, 使用channel.writeUtf8String 写入文件描述符中
• channel.onread: 获取到数据时触发, 跟 channel.writeUtf8String 相对应。通过 JSON.parse 反序列化 message 之后, 调用 handleMessage 进而触发对应事件
• handleMessage: 用来判断是触发 message 事件还是 internalMessage 事件
• target.on('internalMessage'): 针对内部对象做特殊处理，在调用 message 事件

进程间消息传递

• 父进程通过 child.send 发送消息 和 server/socket 句柄对象

• 普通消息直接 JSON.stringify 序列化；对于句柄对象来说，需要先包装成为内部对象

  message = {
  	cmd: 'NODE_HANDLE',
  	type: null,
  	msg: message
  };

  通过 handleConversion.[message.type].send 的方法取出句柄对象对应的 C++ 层面的 TCP 对象，在采用JSON.stringify 序列化

  const handleConversion = {
  	'net.Server': {
      simultaneousAccepts: true,
  
      send(message, server, options) {
        return server._handle;
      },
  
      got(message, handle, emit) {
        const server = new net.Server();
        server.listen(handle, () => {
          emit(server);
        });
      }
    }
  //....
  }

• 最后将序列化后的内部对象和 TCP 对象写入到 IPC 通道中

• 子进程在接收到消息之后，使用 JSON.parse 反序列化消息，如果为内部对象触发 internalMessage 事件

• 检查是否带有 TCP 对象，通过 handleConversion.[message.type].got 得到和父进程一样的句柄对象

• 最后发触发 message 事件传递处理好的消息和句柄对象，子进程通过 process.on 接收

更多node相关知识，请访问：nodejs 教程！

위 내용은 Node의 프로세스 간 통신에 대해 이야기하는 기사의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!