多核机器上的 Deno

Question

在 Node.js 中，有一个 cluster 模块可以利用机器上的所有可用内核，这非常棒，特别是与节点模块 pm2 一起使用时。但我对 Deno 的一些功能非常感兴趣，但我想知道如何在多核机器上最好地运行它。

我知道有些工作人员可以很好地完成特定任务，但对于正常的网络请求来说，多核机器的性能似乎有些浪费？在 Deno 中获得硬件最大可用性和利用率的最佳策略是什么？

我有点担心，如果你只有一个进程在运行，并且有一些 CPU 密集型任务，无论出于何种原因，它都会“阻止”所有其他进来的请求。在 Node.js 中，集群模块可以解决这个问题，因为另一个进程会处理该请求，但我不确定如何在 Deno 中处理这个问题？

我认为您可以在 Deno 中的不同端口上运行多个实例，然后在其前面放置某种负载均衡器，但相比之下，这似乎是一个相当复杂的设置。我还了解到您可以使用某种服务，例如 Deno Deploy 或其他服务，但我已经拥有想要运行它的硬件。

我有哪些替代方案？ 预先感谢您的明智建议和更好的智慧。

Answer 1

在 Deno 中，就像在 Web 浏览器中一样，您应该能够 使用 Web Workers 100% 地利用多核 CPU。

在集群中，您需要一个“管理器”节点（根据需要/适当，它本身也可以是工作器）。以类似的方式，Web Worker API 可用于创建然而，需要有许多敬业的工人。这意味着主线程永远不应该阻塞，因为它可以将所有可能阻塞的任务委托给其工作线程。不会阻塞的任务（例如简单的数据库或其他 I/O 绑定调用）可以像平常一样直接在主线程上完成。

Deno 还支持 navigator.hardwareConcurrency 这样您就可以查询可用的硬件并相应地确定所需的工作人员数量。不过，您可能不需要定义任何限制。从与先前生成的专用工作人员相同的来源生成新的专用工作人员可能足够快，可以按需执行此操作。即便如此，重用专门的工作人员而不是为每个请求生成一个新的工作人员可能是有价值的。

通过可转让对象，可以向工作人员提供大型数据集或从工作人员那里获取大型数据集，而无需复制数据。这与消息传递一起使其变得非常简单委派任务，同时避免复制大型数据集带来的性能瓶颈。

根据您的用例，您还可以使用像 Comlink " 这样的库，它消除了思考 < code>postMessage 并隐藏您正在与工人一起工作的事实。”

例如

ma​​in.ts

import { serve } from "https://deno.land/[email protected]/http/server.ts";

import ComlinkRequestHandler from "./ComlinkRequestHandler.ts";

serve(async function handler(request) {
  const worker = new Worker(new URL("./worker.ts", import.meta.url).href, {
    type: "module",
  });

  const handler = ComlinkRequestHandler.wrap(worker);

  return await handler(request);
});

worker.ts

/// <reference no-default-lib="true"/>
/// <reference lib="deno.worker" />

import ComlinkRequestHandler from "./ComlinkRequestHandler.ts";

ComlinkRequestHandler.expose(async (request) => {
  const body = await request.text();
  return new Response(`Hello to ${request.url}\n\nReceived:\n\n${body}\n`);
});

ComlinkRequestHandler.ts

import * as Comlink from "https://cdn.skypack.dev/[email protected]?dts";

interface RequestMessage extends Omit<RequestInit, "body" | "signal"> {
  url: string;
  headers: Record<string, string>;
  hasBody: boolean;
}

interface ResponseMessage extends ResponseInit {
  headers: Record<string, string>;
  hasBody: boolean;
}

export default class ComlinkRequestHandler {
  #handler: (request: Request) => Promise<Response>;
  #responseBodyReader: ReadableStreamDefaultReader<Uint8Array> | undefined;

  static expose(handler: (request: Request) => Promise<Response>) {
    Comlink.expose(new ComlinkRequestHandler(handler));
  }

  static wrap(worker: Worker) {
    const { handleRequest, nextResponseBodyChunk } =
      Comlink.wrap<ComlinkRequestHandler>(worker);

    return async (request: Request): Promise<Response> => {
      const requestBodyReader = request.body?.getReader();

      const requestMessage: RequestMessage = {
        url: request.url,
        hasBody: requestBodyReader !== undefined,
        cache: request.cache,
        credentials: request.credentials,
        headers: Object.fromEntries(request.headers.entries()),
        integrity: request.integrity,
        keepalive: request.keepalive,
        method: request.method,
        mode: request.mode,
        redirect: request.redirect,
        referrer: request.referrer,
        referrerPolicy: request.referrerPolicy,
      };

      const nextRequestBodyChunk = Comlink.proxy(async () => {
        if (requestBodyReader === undefined) return undefined;
        const { value } = await requestBodyReader.read();
        return value;
      });

      const { hasBody: responseHasBody, ...responseInit } = await handleRequest(
        requestMessage,
        nextRequestBodyChunk
      );

      const responseBodyInit: BodyInit | null = responseHasBody
        ? new ReadableStream({
            start(controller) {
              async function push() {
                const value = await nextResponseBodyChunk();
                if (value === undefined) {
                  controller.close();
                  return;
                }
                controller.enqueue(value);
                push();
              }

              push();
            },
          })
        : null;

      return new Response(responseBodyInit, responseInit);
    };
  }

  constructor(handler: (request: Request) => Promise<Response>) {
    this.#handler = handler;
  }

  async handleRequest(
    { url, hasBody, ...init }: RequestMessage,
    nextRequestBodyChunk: () => Promise<Uint8Array | undefined>
  ): Promise<ResponseMessage> {
    const request = new Request(
      url,
      hasBody
        ? {
            ...init,
            body: new ReadableStream({
              start(controller) {
                async function push() {
                  const value = await nextRequestBodyChunk();
                  if (value === undefined) {
                    controller.close();
                    return;
                  }
                  controller.enqueue(value);
                  push();
                }

                push();
              },
            }),
          }
        : init
    );
    const response = await this.#handler(request);
    this.#responseBodyReader = response.body?.getReader();
    return {
      hasBody: this.#responseBodyReader !== undefined,
      headers: Object.fromEntries(response.headers.entries()),
      status: response.status,
      statusText: response.statusText,
    };
  }

  async nextResponseBodyChunk(): Promise<Uint8Array | undefined> {
    if (this.#responseBodyReader === undefined) return undefined;
    const { value } = await this.#responseBodyReader.read();
    return value;
  }
}

示例用法：

% deno run --allow-net --allow-read main.ts

% curl -X POST --data '{"answer":42}' http://localhost:8000/foo/bar
Hello to http://localhost:8000/foo/bar

Received:

{"answer":42}

可能有更好的方法来做到这一点（例如通过 Comlink.transferHandlers 并注册传输处理程序Request、Response 和/或 ReadableStream），但想法是相同的，并且在正文流式传输时将处理更大的请求或响应有效负载通过消息传递。

Answer 2

这一切都取决于您想推到线程的工作负载。如果您对在主线程上运行的DENO HTTP服务器的内置性能感到满意，但是您需要利用多线程以更有效地创建响应，那么它很简单，如Deno V1.29.4。

HTTP服务器将为您提供异步迭代器 Server ，

import { serve } from "https://deno.land/std/http/server.ts";

const server = serve({ port: 8000 });

您可以使用内置功能 poolmedmap 就像

import { pooledMap } from "https://deno.land/[email protected]/async/pool.ts";

const ress = pooledMap( window.navigator.hardwareConcurrency - 1
                      , server
                      , req => new Promise(v => v(respondWith(req))
                      );

for await (const res of ress) {
  // respond with res
}

westwith 一样，只是一个处理收到的请求并生成响应对象的函数。如果 rectionwith 已经是异步函数，那么您甚至不需要将其包裹在承诺中。

但是，如果您想在单独的Therads上运行多个DENO HTTP服务器，那么这也是可能的，但是您需要一个负载平衡器，例如涉及到头部。在这种情况下，您应该在单独的线程处实例化多个DENO HTTP服务器，并在主线程中接收其重新汇编作为单独的异步迭代器。为了实现这一目标，您可以喜欢的每个线程；

在工作端IE ./服务器/server_800x.ts ;

import { serve } from "https://deno.land/std/http/server.ts";

const server = serve({ port: 800X });
console.log("Listening on http://localhost:800X/");

for await (const req of server) {
  postMessage({ type: "request", req });
}

在主线程上，您可以轻松地将相关的Worker HTTP服务器转换为异步迭代器

async function* server_800X() {
  worker_800X.onmessage = event => {
    if (event.data.type === "request") {
      yield event.data.req;
    }
  };
}

for await (const req of server_800X()) {
  // Handle the request here in the main thread
}

，就像您还应该能够通过使用 muxasynciterators 功能在单个流中，然后由 pool> plimedmap 产生。 异步

const muxedServer = new MuxAsyncIterator<Request>();
muxedServer.add(server_8000);
muxedServer.add(server_8001);
for await (const req of muxedServer) {
  // repond accordingly(*)
}

因此，如果您有2台HTTP服务器在 server_8000.ts 和 server_8001.ts 上工作，那么您可以像显然应该能够产生新的新 通过使用 poolmedmap 从 muxedServer 接收到的请求的线程，如上所示。

（*）如果您选择使用负载平衡器和多个DENO HTTP服务器，则应将特殊标题分配给负载平衡器的请求，并指定已将其转移到的服务器ID。这样，通过检查此磁带标头，您可以从哪个服务器响应任何特定请求。

Answer 3

window.navigator.hardwareconcurrency -1
BlockQuote

deNo使用用于IO操作，垃圾收集和其他实用程序操作，为您提供了更有效的主线程处理。结果，最有效的核心配置的计算可能是非平凡的过程，需要大量的实际测试。在直观级别上，我将为16核服务器创建11-12个工人，但需要其他验证。