钜形不锈钢水箱 发表于 2025-3-1 16:40:46

[肥用云计算] serverless相互调用

前言

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

   具体的配景:bfc 是 listener,afc 是 caller,afc -> bfc。
create listener

listner 具体的方法:

[*]创建 listener bfc,此中 mns trigger 就是监听着 demo eventbridge
[*]创建 eventbridge,取名是 demo eventbridge,同时创建 mns queue,名字是 domo
https://i-blog.csdnimg.cn/img_convert/492c9ca918c645fa94755f76b3b4c3e4.png
⚠️ 开启 eventbridge 中的 base64 解密发送,如下图:
https://i-blog.csdnimg.cn/img_convert/8301c230ee8d6bceca2cdb540e6ce75f.png
同时使用 s deploy 进行部署。
test listener

可以使用本地的 send_mns.py 的代码,进行测试。
# 环境变量
export ALIBABA_CLOUD_ACCESS_KEY_ID="aa"
export ALIBABA_CLOUD_ACCESS_KEY_SECRET="bb"

部门的代码片段:

def create_mns_client():
    """
    创建 MNS 客户端
    """
    # 从环境变量获取阿里云账号 AccessKey
    access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')
    access_key_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')

    if not access_key_id or not access_key_secret:
      raise ValueError("请设置环境变量 ALIYUN_ACCESS_KEY_ID 和 ALIYUN_ACCESS_KEY_SECRET")

    # MNS endpoint格式:https://{AccountId}.mns.{RegionId}.aliyuncs.com
    endpoint = "https://1719759326012690.mns.cn-shenzhen.aliyuncs.com"

    # 创建 MNS 客户端
    return Account(
      endpoint,
      access_key_id,
      access_key_secret
    )

def send_message(queue_name: str, message_body: dict):
    """
    发送消息到 MNS 队列
    @param queue_name: 队列名称
    @param message_body: 消息内容(字典格式)
    """
    try:
      # 获取 MNS 客户端
      account = create_mns_client()
      # 获取队列对象
      queue = account.get_queue(queue_name)

      # 将字典转换为 JSON 字符串
      message_str = json.dumps(message_body)

      # 发送消息
      msg = Message(message_str)
      ret_msg = queue.send_message(msg)

      print(f"消息发送成功!消息ID: {ret_msg.message_id}")
      print(f"消息体: {message_str}")

    except MNSExceptionBase as e:
      print(f"发送消息失败: {str(e)}")
      raise

if __name__ == '__main__':
    # 测试发送消息
    queue_name = "demo"
    test_message = {
      "key1": "command value1",
      "key2": "command value2",
      "timestamp": "2024-01-01 12:00:00"
    }

    send_message(queue_name, test_message)
调用 python mns_send.py,引发测试功能。
观察 listener, s instance list s logs c-67bc1f31-178ea4f7-540e12151f0b --tail
https://i-blog.csdnimg.cn/img_convert/8be00386c20dcd9e2190eb870d0adf41.png
观察日记,发送乐成,符合预期。
页: [1]
查看完整版本: [肥用云计算] serverless相互调用