IRIS 嵌入式 Python 与 Azure Service Bus (ASB) 的用例
概述
三年前，我们开始将 Azure Service Bus（ASB）作为企业消息传递解决方案投入使用。它目前被用于在组织内的众多应用程序之间发布和消费数据。 由于数据流较为复杂，且一个应用程序的数据通常需要被多个应用程序使用，因此“发布者”→“多个订阅者”的模式非常适合。本组织中ASB的日消息量达数千万条，而IRIS平台的日消息量则约为200万至300万条。
ASB 存在的问题
在开始进行 ASB 集成时，我们发现 AMQP 协议并非 IRIS 部署的“开箱即用”选项，因此我们一直在寻找一种替代方案，以便与 ASB 进行通信。
解决方案
我们开发了一个本地 Windows 服务（基于 .NET 和 Swagger），由其负责与 ASB 进行实际通信。该服务安装在与 IRIS 相同的机器上。我们将消息发送到 ASB 时，会先发送给这个本地 Windows 服务（通过 localhost 并调用 REST API），随后由该服务完成后续处理。 该方案运行了几年，效果良好，但监控流量、调试“丢失”的消息以及衡量整体性能却非常困难。事实上，将这个本地 Windows 服务作为“中间人”始终不是最佳架构。
IRIS 嵌入式 Python 早期访问计划 (EAP)
我受邀（或者是我自告奋勇的，记不清了）参与了嵌入式 Python 的早期发布计划（ERP）。这让我非常兴奋，能够测试新功能并向开发团队提供反馈。能够积极参与并影响该产品的开发，这种感觉也特别好。 在此阶段，我们开始测试嵌入式 Python 的 ERP 版本，并决定验证是否能利用嵌入式 Python 来解决“实际”问题。为此选择“IRIS 与 ASB 的直接连接”作为测试案例，是再自然不过的选择。
概念验证（POC）
我们开始编写解决方案的代码，并发现使用 Microsoft ASB 的 Python 库能让我们的工作（和编码）变得轻松许多。 最初阶段是开发一个能够连接到 ASB 中特定主题并接收消息的函数。这一部分完成得相当快（仅用 1 天开发时间），随后我们进入了“发布”（向 ASB 发送）阶段。
又花了几天时间，我们为这些发送和接收函数构建了完整的“封装”：我们实现了以下内容：
- 用于控制消息进出流的规范化“暂存区”
- 一个中央机制，用于存储 ASB 的进出流量，以便进行准确的统计和性能测量
- 一个 CSP 监控页面，用于启用/禁用服务、显示统计数据以及在出现任何问题时发出警报
实现过程
初步配置
在使用 Microsoft ASB Python 库之前，需要将其安装到 ..\python 专用文件夹中。
我们选择使用 ../mge/python/，但您也可以使用任何其他喜欢的文件夹（最初这是一个空文件夹）
这些命令需在以管理员身份运行的 CMD（Windows 系统）中执行，或使用具有管理员权限的用户（Linux 系统）：
..\bin\irispip install --target ..\mgr\python asyncio
..\bin\irispip install --target ..\mgr\python azure.servicebus
接收发往 ASB 的消息（消费）
我们有一个带参数的类方法：
- topiId - ASB 被划分为多个主题，您可以从中消费消息
- subscriptionName - Azure 订阅名称
- connectionString - 用于连接您已订阅的主题
请注意，我们使用 [ Language = python ] 来表示此类方法是用 Python 编写的 (!)
要使用 ASB 库，我们首先需要导入 ServiceBusClient 和 ServiceBusSubQueue 库：
为了能够与 IRIS 进行交互（例如运行代码），我们还需要：
此时，我们就可以使用 ASB 库了：
with ServiceBusClient.from_connection_string(connectionString) as client:
with client.get_subscription_receiver(topicName, subscriptionName, max_wait_time=1, sub_queue=subQueue) as receiver:
for msg in receiver:
此时，我们已获得一个 Python 对象“msg”（流），可以将其（当然是以流的形式）传递给 IRIS，并直接存储到数据库中：
向 ASB 发送消息（发布）
我们有一个带参数的类方法：
- topicName - 我们要发布到的主题（此处需传入名称而非 ID）
- connectionString - 用于连接您已订阅的主题
- JSONmessage - 我们要发送（发布）的消息
请注意，我们使用 [ Language = python ] 来表示该类方法是用 Python 编写的 (!)
要使用 ASB 库，我们首先需要导入 ServiceBusClient 和 ServiceBusMessage 库：
随后使用 ASB 库就非常简单了：
try:
result=1
with ServiceBusClient.from_connection_string(connectionString) as client:
with client.get_queue_sender(topicName) as sender:
single_message = ServiceBusMessage(JSONmessage)
sender.send_messages(single_message)
except Exception as e:
print(e)
result=0
return result
使用直接 ASB 连接的优势
- 比使用“Windows 本地服务”这种“旧方案”快得多
- 便于监控、收集统计数据及排查问题
- 淘汰“中间人”（Windows 本地服务）可减少潜在的“单点故障”
- 能够自动管理任何主题的“死信”，并让 ASB 将这些消息重新发送至订阅者的主题
鸣谢
我要感谢 @David.Satorres6134 （我们的 IRIS 高级开发人员），感谢他在 设计、编码和测试方面 做出的巨大 贡献。 没有他的帮助，这个项目就无法实现。