解决Python虚拟环境中WebSocket回调函数不执行的问题-Python教程-PHP中文网

解决Python虚拟环境中WebSocket回调函数不执行的问题

本文探讨了Python虚拟环境中WebSocket on_ticks 回调函数不执行的常见问题。核心原因在于WebSocket连接在订阅后被过早关闭，或主线程在异步任务完成前退出。解决方案是引入阻塞操作（如 input() 或 time.sleep()）来维持连接的活跃状态和主线程的生命周期，确保回调函数有足够时间被触发。文章提供了示例代码和针对生产环境的建议。

问题描述与背景

在使用python进行量化交易或实时数据处理时，经常需要通过websocket连接接收实时行情数据。开发者可能遇到这样的情况：在本地开发环境中，on_ticks 等websocket回调函数能够正常接收并处理数据，但在激活的python虚拟环境中运行相同的代码时，回调函数却无法被触发，没有任何数据输出。

以下是原始代码示例，它是一个Django管理命令，用于连接BreezeConnect API并订阅行情数据：

import time
from typing import Any
from django.core.management.base import BaseCommand
from breezeconnect import BreezeConnect

class Command(BaseCommand):
    help = 'Connects to Breeze API and subscribes to market feeds.'

    def handle(self, *args: Any, **options: Any):
        api_key = "YOUR_API_KEY"
        api_secret = "YOUR_API_SECRET"
        session_token = "YOUR_SESSION_TOKEN"

        print("Connecting to Breeze")
        breeze = BreezeConnect(api_key=api_key)

        breeze.generate_session(api_secret=api_secret, session_token=session_token)
        print("Session generated successfully")

        breeze.ws_connect()
        print("WebSocket connected successfully")

        def on_ticks(ticks):
            print("Ticks: {}".format(ticks))

        breeze.on_ticks = on_ticks

        breeze.subscribe_feeds(exchange_code="NFO", stock_code="ADAENT", product_type="options",
                                expiry_date="28-Dec-2023", strike_price="3000", right="Call",
                                get_exchange_quotes=True, get_market_depth=False)

        print("Subscribed to ADAENT options")

        breeze.ws_disconnect()
        print("Disconnected from WebSocket")

登录后复制

当此代码在虚拟环境中通过 python3 manage.py your_command_name 运行时，on_ticks 函数不会被调用，也没有数据打印。

根本原因分析

要理解为何回调不执行，我们需要关注WebSocket连接的异步特性以及程序的主线程生命周期。

WebSocket的异步性质： breeze.ws_connect() 方法会建立一个WebSocket连接。通常，这个连接会在后台（可能通过一个独立的线程或内部事件循环）维护，并等待服务器推送数据。当数据到达时，分配给 breeze.on_ticks 的回调函数才会被触发。这是一个异步过程，即主程序流不会等待数据到达。

立即学习“Python免费学习笔记（深入）”；
过早的连接关闭： 仔细观察上述代码，在调用 breeze.subscribe_feeds() 订阅数据之后，紧接着就调用了 breeze.ws_disconnect()。这意味着WebSocket连接在订阅请求发出后几乎立即就被关闭了。由于网络延迟、订阅确认时间以及数据推送的时间，在连接被关闭之前，几乎没有时间让任何行情数据到达并触发 on_ticks 回调。
主线程的生命周期： 即使没有 breeze.ws_disconnect() 调用，如果主线程在启动WebSocket连接和订阅后立即退出（例如，Django管理命令执行完毕），那么整个Python进程可能会被终止。在某些操作系统或Python版本中，主线程的退出会导致所有后台线程或异步任务被强制终止，从而阻止任何待处理的回调函数被执行。虚拟环境可能对进程生命周期的管理有不同的行为，导致这种现象在本地环境和虚拟环境之间表现不一致。

因此，问题的核心在于没有给WebSocket连接足够的时间来接收数据并执行回调，无论是由于显式地过早断开连接，还是由于主线程过早退出导致整个进程终止。

阿里云-虚拟数字人

阿里云-虚拟数字人是什么？ ...

查看详情

解决方案：维持连接和主线程活跃

为了确保 on_ticks 回调函数能够被触发，我们需要在订阅数据后，让程序保持活跃状态，并维持WebSocket连接。

1. 使用 input() 阻塞主线程 (适用于交互式调试)

最简单直接的方法是在程序结束前添加一个阻塞操作，例如 input()。这会暂停程序的执行，直到用户按下任意键，从而为WebSocket连接提供足够的时间来接收数据。

import time
from typing import Any
from django.core.management.base import BaseCommand
from breezeconnect import BreezeConnect

class Command(BaseCommand):
    help = 'Connects to Breeze API and subscribes to market feeds.'

    def handle(self, *args: Any, **options: Any):
        api_key = "YOUR_API_KEY"
        api_secret = "YOUR_API_SECRET"
        session_token = "YOUR_SESSION_TOKEN"

        print("Connecting to Breeze")
        breeze = BreezeConnect(api_key=api_key)

        breeze.generate_session(api_secret=api_secret, session_token=session_token)
        print("Session generated successfully")

        breeze.ws_connect()
        print("WebSocket connected successfully")

        def on_ticks(ticks):
            print("Ticks: {}".format(ticks))

        breeze.on_ticks = on_ticks

        breeze.subscribe_feeds(exchange_code="NFO", stock_code="ADAENT", product_type="options",
                                expiry_date="28-Dec-2023", strike_price="3000", right="Call",
                                get_exchange_quotes=True, get_market_depth=False)

        print("Subscribed to ADAENT options. Waiting for ticks...")

        # 关键改动：添加一个阻塞调用，保持程序活跃
        # 这将延迟 ws_disconnect() 的执行，直到用户按下回车
        try:
            input("Press Enter to disconnect and exit...\n")
        except KeyboardInterrupt:
            print("\nExiting gracefully...")
        finally:
            breeze.ws_disconnect()
            print("Disconnected from WebSocket")

登录后复制

解释： 通过在 breeze.subscribe_feeds() 之后和 breeze.ws_disconnect() 之前插入 input()，我们有效地阻止了主线程立即退出或关闭连接。这使得WebSocket连接能够持续活跃，有足够的时间接收行情数据并触发 on_ticks 回调。当用户按下回车键时，input() 完成，程序继续执行 ws_disconnect() 并最终退出。

2. 使用 time.sleep() 保持活跃 (适用于非交互式场景)

如果你的Django管理命令需要作为后台服务运行，而不是交互式地等待用户输入，那么 input() 就不适用。在这种情况下，你可以使用一个循环结合 time.sleep() 来让程序保持活跃。

import time
from typing import Any
from django.core.management.base import BaseCommand
from breezeconnect import BreezeConnect

class Command(BaseCommand):
    help = 'Connects to Breeze API and subscribes to market feeds.'

    def handle(self, *args: Any, **options: Any):
        api_key = "YOUR_API_KEY"
        api_secret = "YOUR_API_SECRET"
        session_token = "YOUR_SESSION_TOKEN"

        print("Connecting to Breeze")
        breeze = BreezeConnect(api_key=api_key)

        breeze.generate_session(api_secret=api_secret, session_token=session_token)
        print("Session generated successfully")

        breeze.ws_connect()
        print("WebSocket connected successfully")

        def on_ticks(ticks):
            print("Ticks: {}".format(ticks))
            # 可以在这里添加数据处理逻辑，例如存储到数据库

        breeze.on_ticks = on_ticks

        breeze.subscribe_feeds(exchange_code="NFO", stock_code="ADAENT", product_type="options",
                                expiry_date="28-Dec-2023", strike_price="3000", right="Call",
                                get_exchange_quotes=True, get_market_depth=False)

        print("Subscribed to ADAENT options. Running indefinitely, press Ctrl+C to stop.")

        # 关键改动：使用循环和 time.sleep() 保持程序活跃
        try:
            while True:
                # 可以在这里执行其他非阻塞任务，或者仅仅是等待
                time.sleep(1) # 每秒检查一次，保持主线程活跃
        except KeyboardInterrupt:
            print("\nExiting gracefully...")
        finally:
            breeze.ws_disconnect()
            print("Disconnected from WebSocket")

登录后复制

解释：while True: time.sleep(1) 结构创建了一个无限循环，使主线程每秒暂停一秒。这确保了主线程不会退出，从而允许WebSocket连接在后台持续运行并触发 on_ticks 回调。try...except KeyboardInterrupt 块用于捕获 Ctrl+C 信号，实现程序的优雅退出，并在退出前断开WebSocket连接。

生产环境考量与最佳实践

对于生产环境中的长时运行服务，仅仅依靠 input() 或 time.sleep() 循环可能不够健壮。

进程管理： 对于需要持续运行的Django管理命令，应将其配置为守护进程，并使用专业的进程管理器（如 Supervisor, systemd 或 Docker）来管理其生命周期。这些工具可以在程序崩溃时自动重启，并提供日志管理等功能。
异步框架集成： 如果 BreezeConnect 库支持 asyncio 或其他异步框架，将其集成到Django的异步任务或Celery worker中会是更优雅的解决方案。这样可以更好地管理并发和资源。
错误处理与重连机制： 在实际应用中，WebSocket连接可能会断开。on_error 和 on_close 回调应该被实现，以处理连接错误和断开事件，并尝试自动重连。
日志记录： 详细的日志记录对于诊断生产环境中的问题至关重要。将 on_ticks 中的数据打印替换为结构化的日志输出。
资源管理： 确保在程序退出时（无论是正常退出还是异常退出），WebSocket连接都能被正确关闭，释放资源。try...finally 块是实现这一点的有效方式。

总结

当Python虚拟环境中WebSocket回调函数不执行时，最常见的原因是程序没有给异步任务足够的时间来完成。这通常表现为WebSocket连接在订阅后被过早关闭，或者主线程在后台任务完成前退出。通过在关键异步操作后引入阻塞机制（如 input() 或 time.sleep() 循环），可以有效地保持连接活跃并维持主线程生命周期，从而确保回调函数能够正常触发。在部署到生产环境时，应进一步考虑使用专业的进程管理工具和更健壮的错误处理及重连机制。

以上就是解决Python虚拟环境中WebSocket回调函数不执行的问题的详细内容，更多请关注php中文网其它相关文章！