如何使用 Django Channels 动态创建和关闭无限运行的 Celery 任务

如何解决如何使用 Django Channels 动态创建和关闭无限运行的 Celery 任务

我正在创建一个加密货币应用程序，用户可以在其中获取货币对的实时数据（例如“BTC/USD”）。我正在尝试建立一个系统，订阅同一对的用户可以共享 API 调用。

到目前为止，我可以动态创建 websocket 连接并将用户添加到组中。

# routing.py
URLRouter([
        path('ws/data/<pair>/',PriceFeedsConsumer.as_asgi()),])

# consumers.py
class PriceFeedsConsumer(AsyncWebsocketConsumer):

async def connect(self):
    self.pair = self.scope['url_route']['kwargs']['pair']
    self.pair_group_name = 'pair_group_%s' % self.pair

    await self.channel_layer.group_add(
        self.pair_group_name,self.channel_name
    )

    await self.accept()

...

我已将 redis 设置为代理服务，将 celery 设置为任务管理器系统。 我知道如何将 celery 任务的数据发送到消费者组，但我想在创建组时开始任务，并在丢弃组时最终关闭任务。 这似乎是管理 API 调用的最有效方式。

是否可以通过创建 django 组来初始化 celery 任务？这样做时，将一个参数（在本例中为对）传递给 celery 任务？

或者，这是解决此类问题的正确架构吗？