python - Asyncio 循环等待事件

Question

我的问题是我有用于事件监听的脚本。连接到服务器后，事件监听开​​始循环。因此，在函数 main 中连接服务器后，我创建了循环

loop = asyncio.get_event_loop()
        asyncio.ensure_future(monitor(ts3conn, dbconn))
        loop.run_forever()

现在，当我收到任何事件时，检查 if 语句，如果有 True，我必须等待函数或创建新任务？我想要异步 3 个函数，1 个主要函数，一直监听，以及当有人写入消息/加入 channel 时将创建的附加内容，在此脚本中，当 30 个用户加入 channel 777 时，asyncio 仍然无法工作(如果 777 == int(event ['ctid']):) 同时有人会加入 channel 904 (如果 904 == int(event['ctid']):)，最后一个人必须等到 30 个用户得到服务(我希望你理解)

我的代码:

import ts3
import time
import logging
import json
import pymysql.cursors
import pymysql
import asyncio
from logging.handlers import RotatingFileHandler

def main():
    with ts3.query.TS3ServerConnection(URI) as ts3conn:
        # connect to server instance, update name and go to specific channel
        ts3conn.exec_("use", sid=SID)
        ts3conn.exec_("clientupdate", client_nickname=CLIENT_NAME)
        myclid = ts3conn.exec_("whoami")[0]["client_id"]
        ts3conn.exec_("clientmove", clid=myclid, cid=JOIN_CHANNEL_ID)
        ts3conn.exec_("servernotifyregister", event="server")
        ts3conn.exec_("servernotifyregister", event="channel", id=0)
        ts3conn.exec_("servernotifyregister", event="textprivate")
        dbconn = pymysql.connect(host='localhost', user='root', password='', db='teamspeak')

        loop = asyncio.get_event_loop()
        asyncio.ensure_future(monitor(ts3conn, dbconn))
        loop.run_forever()

# Function handling the events and initiating activity logs
async def monitor(ts3conn, dbconn):
    # register for all events in server wide chat
    ts3conn.exec_("servernotifyregister", event="server")
    ts3conn.exec_("servernotifyregister", event="channel", id=0)
    ts3conn.exec_("servernotifyregister", event="textprivate")
    ts3conn.send_keepalive()
    while True:
        try:
            event = ts3conn.wait_for_event(timeout=10)[0]
        except ts3.query.TS3TimeoutError:
            ts3conn.send_keepalive()
        else:
            await asyncio.sleep(0.001)
            print(event)
            # ============= IF JOIN CHANNEL ===================
            if "ctid" in event.keys() and "clid" in event.keys() and int(event['ctid']) != 0:
                if 777 == int(event['ctid']):
                    asyncio.create_task(first(ts3conn, dbconn, event['clid']))
                    #ts3conn.exec_("clientkick", reasonid=4, clid=event['clid'])
                if 904 == int(event['ctid']):
                    asyncio.create_task(second(ts3conn, dbconn, event['clid']))
                    #ts3conn.exec_("clientkick", reasonid=4, clid=event['clid'])
            # ============= IF SEND MSG ===================
            if "msg" in event.keys() and "invokeruid" in event.keys() and 'serveradmin' not in str(event['invokeruid']):
                if event['msg'] == "!info":
                    print("info")
                    asyncio.create_task(first(ts3conn, dbconn, event['invokerid']))


async def first(ts3conn, dbconn, uid):
    try:
        print("first")
        user = ts3conn.exec_("clientinfo", clid=uid)
        if any(i in user[0]['client_servergroups'] for i in REG):
            try:
                sql = "SELECT * FROM users WHERE uid=%s"
                cursor = dbconn.cursor()
                cursor.execute(sql, (user[0]['client_unique_identifier']))
                c = cursor.fetchone()
                ts3conn.exec_("sendtextmessage", targetmode="1", target=uid, msg=f"register: {c}")
            except KeyError as e:
                print(e)
        else:
            ts3conn.exec_("sendtextmessage", targetmode="1", target=uid, msg=f"not register")
    except KeyError as e:
        print(f"keyerror: {e}")


async def second(ts3conn, dbconn, uid):
    try:
        user = ts3conn.exec_("clientinfo", clid=uid)
        if any(i in user[0]['client_servergroups'] for i in REG):
            try:
                sql = "SELECT * FROM users WHERE uid=%s"
                cursor = dbconn.cursor()
                cursor.execute(sql, (user[0]['client_unique_identifier']))
                c = cursor.fetchone()
                ts3conn.exec_("sendtextmessage", targetmode="1", target=uid, msg=f"1 out: {c}")
            except KeyError as e:
                print(e)
        else:
            ts3conn.exec_("sendtextmessage", targetmode="1", target=uid, msg=f"321123321321132")
    except KeyError as e:
        print(f"keyerror: {e}")



if __name__ == "__main__":
    with open('config.json') as config_file:
        config = json.load(config_file)

    try:
        SQLDATABASE = config["sqldatabase"]
        DATABASE = config["sqldatabase"]
        URI = config["uri"]
        SID = config["sid"]
        CLIENT_NAME = config["client_name"]
        JOIN_CHANNEL_ID = config["join_channel_id"]
        REG = config['zarejeya']

        if config["log_level"] == "CRITICAL":
            LOG_LEVEL = logging.CRITICAL
        elif config["log_level"] == "ERROR":
            LOG_LEVEL = logging.ERROR
        elif config["log_level"] == "WARNING":
            LOG_LEVEL = logging.WARNING
        elif config["log_level"] == "INFO":
            LOG_LEVEL = logging.INFO
        elif config["log_level"] == "DEBUG":
            LOG_LEVEL = logging.DEBUG
        else:
            LOG_LEVEL = logging.NOTSET
    except:
        print("Error parsing config")
        raise

    log_formatter = logging.Formatter("%(asctime)s - %(funcName)s - %(levelname)s - %(message)s")
    log_handler = RotatingFileHandler("ts3bot.log", mode='a', maxBytes=50 * 1024 * 1024, backupCount=2)
    log_handler.setFormatter(log_formatter)
    log_handler.setLevel(LOG_LEVEL)
    # noinspection PyRedeclaration
    logger = logging.getLogger("root")
    logger.setLevel(LOG_LEVEL)
    logger.addHandler(log_handler)

    while True:
        try:
            main()
        except Exception:
            logger.exception("Exception occurred and connection is closed")
            logger.info("Trying to restart in 30s")
            time.sleep(30)

我需要这样的不和谐机器人: https://tutorials.botsfloor.com/a-discord-bot-with-asyncio-359a2c99e256但我无法到达这里...

Answer 1

调用 await() 将使当前协程暂停，并且只有在返回等待的 coro 时才会恢复。

在您的代码中，monitor 协程将串行处理传入的事件。因此，如果发生任何事件，监视器将等待 first()/second() coros，这将使monitor coro再次暂停...所以这是同步的..异步没有真正的好处..相反它使性能更差..

事件处理协程，first()和second()可以转换为任务。您可以使用asyncio.create_task() 将协程包装到任务中。这将被安排在轮到它时在同一事件循环中执行。并且在任务执行时调用 coro(int this case, monitor()) 不会被挂起。任务本身是一个 awaitable，因此在任务将使被调用者暂停。

也许，您可能想保存创建的任务(在 create_task() 期间返回)以供以后的操作使用。例如停止它、等待它......

await task # To await for the task to complete, suspends to current coro.

如果你想取消一个事件的任务，那么该任务将抛出一个CancelledError，这个错误必须妥善处理。

示例:

async def stop_task(task):
    if not task:
        return
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass