如何在AWS API Gateway的$connect Lambda中向新WebSocket连接发消息？

你的推测是对的：在$connect阶段同步调用PostToConnection会失败，因为此时API Gateway还未完成WebSocket的握手流程，连接尚未被标记为活跃状态。要实现连接完成后给用户发送消息，需要通过异步触发的方式延迟消息发送操作。

可行方案：异步触发消息发送 #

核心思路是让$connect的Lambda先返回200确认连接，再通过其他服务（如SQS、Lambda异步调用）触发消息发送逻辑，确保连接已完全建立。

方案1：使用SQS中转消息 #

通过SQS队列暂存连接信息，由独立的Lambda监听队列并发送消息，这是最可靠的异步方式。

步骤1：修改$connect Lambda（发送连接信息到SQS）

import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";

export const handler = async (event) => {
  const domain = event.requestContext.domainName;
  const stage = event.requestContext.stage;
  const connectionId = event.requestContext.connectionId;
  
  // 初始化SQS客户端
  const sqsClient = new SQSClient({ region: "你的AWS区域" });
  const queueUrl = "你的SQS队列URL";
  
  try {
    // 将连接信息发送到SQS
    await sqsClient.send(new SendMessageCommand({
      QueueUrl: queueUrl,
      MessageBody: JSON.stringify({ domain, stage, connectionId })
    }));
  } catch (sqsError) {
    console.error("发送到SQS失败:", sqsError);
  }

  // 立即返回200，完成连接握手
  return { statusCode: 200 };
};

步骤2：创建消息发送Lambda（监听SQS并发送消息）

import {
  ApiGatewayManagementApiClient,
  PostToConnectionCommand,
} from "@aws-sdk/client-apigatewaymanagementapi";

export const handler = async (event) => {
  for (const record of event.Records) {
    const { domain, stage, connectionId } = JSON.parse(record.body);
    const callbackUrl = `https://${domain}/${stage}`;
    const client = new ApiGatewayManagementApiClient({ endpoint: callbackUrl });

    const requestParams = {
      ConnectionId: connectionId,
      Data: JSON.stringify({ data: "Hello World" }),
    };

    try {
      await client.send(new PostToConnectionCommand(requestParams));
    } catch (error) {
      console.error("发送消息失败:", error);
      // 410错误表示连接已关闭，可跳过清理
      if (error.statusCode === 410) {
        console.log(`连接${connectionId}已关闭，忽略`);
      }
    }
  }
  return { statusCode: 200 };
};

权限配置

• 给$connect Lambda的IAM角色添加sqs:SendMessage权限，目标为你的SQS队列ARN。
• 给消息发送Lambda的IAM角色添加：
  • apigateway:POST_TO_CONNECTION权限，资源为arn:aws:apigateway:你的区域::/connections/*
  • SQS的sqs:ReceiveMessage、sqs:DeleteMessage、sqs:GetQueueAttributes权限，目标为你的SQS队列ARN。

方案2：Lambda异步调用 #

直接在$connect Lambda中异步调用另一个Lambda执行消息发送逻辑，无需额外中间件。

修改$connect Lambda（触发异步Lambda）

import { LambdaClient, InvokeCommand } from "@aws-sdk/client-lambda";

export const handler = async (event) => {
  const domain = event.requestContext.domainName;
  const stage = event.requestContext.stage;
  const connectionId = event.requestContext.connectionId;
  
  const lambdaClient = new LambdaClient({ region: "你的AWS区域" });
  const sendMessageLambdaName = "你的消息发送Lambda名称";
  
  try {
    // 异步调用消息发送Lambda
    await lambdaClient.send(new InvokeCommand({
      FunctionName: sendMessageLambdaName,
      InvocationType: "Event", // 标记为异步执行
      Payload: Buffer.from(JSON.stringify({ domain, stage, connectionId }))
    }));
  } catch (lambdaError) {
    console.error("触发异步Lambda失败:", lambdaError);
  }

  return { statusCode: 200 };
};

消息发送Lambda代码

import {
  ApiGatewayManagementApiClient,
  PostToConnectionCommand,
} from "@aws-sdk/client-apigatewaymanagementapi";

export const handler = async (event) => {
  const { domain, stage, connectionId } = JSON.parse(event.toString());
  const callbackUrl = `https://${domain}/${stage}`;
  const client = new ApiGatewayManagementApiClient({ endpoint: callbackUrl });

  const requestParams = {
    ConnectionId: connectionId,
    Data: JSON.stringify({ data: "Hello World" }),
  };

  try {
    await client.send(new PostToConnectionCommand(requestParams));
  } catch (error) {
    console.error("发送消息失败:", error);
    if (error.statusCode === 410) {
      console.log(`连接${connectionId}已关闭`);
    }
  }
  
  return { statusCode: 200 };
};

权限配置

• 给$connect Lambda的IAM角色添加lambda:InvokeFunction权限，目标为消息发送Lambda的ARN。
• 给消息发送Lambda的IAM角色添加apigateway:POST_TO_CONNECTION权限，资源为arn:aws:apigateway:你的区域::/connections/*。

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