使用aigraph创建后台进程 [英] Create a background process using aiogram
本文介绍了使用aigraph创建后台进程的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我正在开发一个使用aigraph的电报机器人,我正在尝试发送加密货币的价格警报。我遇到的问题是,我不确定如何启动作为后台、非阻塞线程的函数,然后继续启动调度程序。我知道如何使用标准的同步电报机器人来做到这一点,但我不知道我应该如何处理aigraph。我读到我可以使用dp.loop.create_task
here,但这引发了一个错误Nonetype has no attribute create_task
。下面是我尝试用来执行这些线程的代码:
print('Starting watchlist process, this needs to run as a non blocking daemon...')
dp.loop.create_task(wl.start_process())
print('Starting broadcaster, this needs to run as a non blocking daemon ... ')
dp.loop.create_task(broadcaster())
print('Starting the bot ...')
executor.start_polling(dp, skip_updates=True)
我只需要在后台运行wl.start_process
和broadcaster
函数。我如何完成此操作?
这是start_process
:
async def start_process(self):
"""
Start the watchlist process.
:return:
"""
threading.Thread(target=self.do_schedule).start()
await self.loop_check_watchlist()
这是broadcaster
:
async def broadcaster():
count = 0
while True:
uid_alerts = que.__next__()
if uid_alerts:
for i in uid_alerts:
uid = i[0]
alert = i[1]
try:
if await send_message(uid, alert):
count += 1
await asyncio.sleep(.05) # 20 messages per second (Limit: 30 messages per second)
finally:
log.info(f"{count} messages successful sent.")
推荐答案
我为您创建了以下工作示例,它演示了如何在机器人启动时创建后台任务,以及如何使用用户命令创建后台任务。
from aiogram import Dispatcher, executor, Bot
from aiogram import types
import asyncio
import logging
TOKEN = "your token!"
async def background_on_start() -> None:
"""background task which is created when bot starts"""
while True:
await asyncio.sleep(5)
print("Hello World!")
async def background_on_action() -> None:
"""background task which is created when user asked"""
for _ in range(20):
await asyncio.sleep(3)
print("Action!")
async def background_task_creator(message: types.Message) -> None:
"""Creates background tasks"""
asyncio.create_task(background_on_action())
await message.reply("Another one background task create")
async def on_bot_start_up(dispatcher: Dispatcher) -> None:
"""List of actions which should be done before bot start"""
asyncio.create_task(background_on_start()) # creates background task
def create_bot_factory() -> None:
"""Creates and starts the bot"""
dp = Dispatcher(Bot(token=TOKEN))
# bot endpoints block:
dp.register_message_handler(
background_task_creator,
commands=['start']
)
# start bot
executor.start_polling(dp, skip_updates=True, on_startup=on_bot_start_up)
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
create_bot_factory()
这篇关于使用aigraph创建后台进程的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文