IT评测·应用市场-qidao123.com技术社区

标题: [肥用云计算] serverless相互调用 [打印本页]
作者: 钜形不锈钢水箱    时间: 2025-3-1 16:40
标题: [肥用云计算] serverless相互调用
前言

在Serverless架构广泛应用的当代软件开发配景下，服务间的通讯已成为一项广泛需求。本文旨在探讨三种紧张的服务间通讯模式：1) 消息队列（Queue），2) HTTP协议，3) 内部网络函数调用（FC Call）。我们将详细先容每种方法的特点、实用场景及实现方式。
queue

   具体的配景：bfc 是 listener，afc 是 caller，afc -> bfc。
  create listener

listner 具体的方法：

⚠️ 开启 eventbridge 中的 base64 解密发送，如下图：

同时使用 s deploy 进行部署。
test listener

可以使用本地的 send_mns.py 的代码，进行测试。
  1. # 环境变量
  2. export ALIBABA_CLOUD_ACCESS_KEY_ID="aa"
  3. export ALIBABA_CLOUD_ACCESS_KEY_SECRET="bb"
复制代码
部门的代码片段：
  2. def create_mns_client():
  3.     """
  4.     创建 MNS 客户端
  5.     """
  6.     # 从环境变量获取阿里云账号 AccessKey
  7.     access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')
  8.     access_key_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
  10.     if not access_key_id or not access_key_secret:
  11.         raise ValueError("请设置环境变量 ALIYUN_ACCESS_KEY_ID 和 ALIYUN_ACCESS_KEY_SECRET")
  13.     # MNS endpoint格式：https://{AccountId}.mns.{RegionId}.aliyuncs.com
  14.     endpoint = "https://1719759326012690.mns.cn-shenzhen.aliyuncs.com"
  16.     # 创建 MNS 客户端
  17.     return Account(
  18.         endpoint,
  19.         access_key_id,
  20.         access_key_secret
  21.     )
  23. def send_message(queue_name: str, message_body: dict):
  24.     """
  25.     发送消息到 MNS 队列
  26.     @param queue_name: 队列名称
  27.     @param message_body: 消息内容（字典格式）
  28.     """
  29.     try:
  30.         # 获取 MNS 客户端
  31.         account = create_mns_client()
  32.         # 获取队列对象
  33.         queue = account.get_queue(queue_name)
  35.         # 将字典转换为 JSON 字符串
  36.         message_str = json.dumps(message_body)
  38.         # 发送消息
  39.         msg = Message(message_str)
  40.         ret_msg = queue.send_message(msg)
  42.         print(f"消息发送成功！消息ID: {ret_msg.message_id}")
  43.         print(f"消息体: {message_str}")
  45.     except MNSExceptionBase as e:
  46.         print(f"发送消息失败: {str(e)}")
  47.         raise
  49. if __name__ == '__main__':
  50.     # 测试发送消息
  51.     queue_name = "demo"
  52.     test_message = {
  53.         "key1": "command value1",
  54.         "key2": "command value2",
  55.         "timestamp": "2024-01-01 12:00:00"
  56.     }
  58.     send_message(queue_name, test_message)
复制代码
调用 python mns_send.py，引发测试功能。
观察 listener， s instance list s logs c-67bc1f31-178ea4f7-540e12151f0b --tail

观察日记，发送乐成，符合预期。




欢迎光临 IT评测·应用市场-qidao123.com技术社区 (https://dis.qidao123.com/) Powered by Discuz! X3.4