AWS API Gateway websockets 聊天示例是完全异步的吗？

Question

我正在关注最近宣布的 AWS API Gateway websockets 支持 -

https://aws.amazon.com/blogs/compute/announcing-websocket-apis-in-amazon-api-gateway/

他们有一个示例聊天服务器示例 -

https://github.com/aws-samples/simple-websockets-chat-app/blob/master/sendmessage/app.js

我正在运行，非常好。

如果您发送消息，sendmessage Lambda 会通过以下方式将该消息广播给所有连接的用户 -

// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: MIT-0

const AWS = require('aws-sdk');

const ddb = new AWS.DynamoDB.DocumentClient({ apiVersion: '2012-08-10' });

const { TABLE_NAME } = process.env;

exports.handler = async (event, context) => {
  let connectionData;

  try {
    connectionData = await ddb.scan({ TableName: TABLE_NAME, ProjectionExpression: 'connectionId' }).promise();
  } catch (e) {
    return { statusCode: 500, body: e.stack };
  }

  const apigwManagementApi = new AWS.ApiGatewayManagementApi({
    apiVersion: '2018-11-29',
    endpoint: event.requestContext.domainName + '/' + event.requestContext.stage
  });

  const postData = JSON.parse(event.body).data;

  const postCalls = connectionData.Items.map(async ({ connectionId }) => {
    try {
      await apigwManagementApi.postToConnection({ ConnectionId: connectionId, Data: postData }).promise();
    } catch (e) {
      if (e.statusCode === 410) {
        console.log(`Found stale connection, deleting ${connectionId}`);
        await ddb.delete({ TableName: TABLE_NAME, Key: { connectionId } }).promise();
      } else {
        throw e;
      }
    }
  });

  try {
    await Promise.all(postCalls);
  } catch (e) {
    return { statusCode: 500, body: e.stack };
  }

  return { statusCode: 200, body: 'Data sent.' };
};

现在不幸的是（幸运的是？？）我有 Python/Erlang 背景，而不是 Javascript/nodejs。所以我可以看到其中一些在做什么，即迭代 DynamoDB 表中的连接并将响应发送到每个连接。通过使用async 和await 关键字，它也看起来 异步工作，我猜这是Promises。但我不能确定这是异步执行的，这让我很担心.. 如果我有一个 Lambda 迭代大量连接并进行同步调用，那是行不通的.

所以 - 特别是关于这部分代码 -

 const postCalls = connectionData.Items.map(async ({ connectionId }) => {
    try {
      await apigwManagementApi.postToConnection({ ConnectionId: connectionId, Data: postData }).promise();
    } catch (e) {
      if (e.statusCode === 410) {
        console.log(`Found stale connection, deleting ${connectionId}`);
        await ddb.delete({ TableName: TABLE_NAME, Key: { connectionId } }).promise();
      } else {
        throw e;
      }
    }
  });

我可以确定这是以完全异步的方式将postData 发布到所有连接吗？我是否需要担心单个 Lambda 可能会将消息推送到数千个客户端？

Answer 1

它不被称为 simple-websockets-chat-app :)

通过阅读代码，它完全符合您的要求。只有一个 lambda 实例会向所有连接触发消息。

但它是一个聊天室，有成千上万的用户很常见吗？

通过使用 async 和 await 关键字，我猜是 Promises

是的，它以异步方式工作，但 lambda 仍将执行，直到所有消息都发送完毕。

---

关于

  await apigwManagementApi.postToConnection({ ConnectionId: connectionId, Data: postData }).promise();

它对 Lambda 本身的压力很小，一次只进行一个远程调用。并且是异步的，它不会等待响应，而是继续发送越来越多的数据。

（旁注 - 在这种情况下，我会使用 forEach 而不是 map）

一个荒谬的高流量聊天室的解决方案是：

• 在“DispatcherLambda”中，将来自 TABLE_NAME 的扫描拆分为多个块，将它们发送到 SQS 队列并为其订阅 Lambda。这样会有多个容器执行该代码（根据您决定的粒度）