Python异步队列接收TikTok评论一段时间后停止填充问题排查

我使用TikTokLive API编写Python异步程序，通过异步函数将评论添加至asyncio.Queue，再用另一个异步函数取出打印。目前遇到的问题是：队列在接收并打印几条评论后，就不再填充新内容，持续输出“Waiting for new Comments...”。尝试使用asyncio.create_task分别运行客户端启动任务与打印任务，问题依旧；直接在on_comment函数中打印评论可正常执行，但无法通过队列复用评论数据。

原代码 #

import asyncio
import time
from asyncio import Queue
from TikTokLive import TikTokLiveClient
from TikTokLive.types.events import CommentEvent, ConnectEvent

#/Users/sunilpc/Documents/Python/tiktokTest.py

# Instantiate the client with the user's username
client: TikTokLiveClient = TikTokLiveClient(unique_id="@territorialwarfr")

# Comment Holders
commentQueue = asyncio.Queue()

# Define how you want to handle specific events via decorator
@client.on("connect")
async def on_connect(_: ConnectEvent):
    print("Connected to Room ID:", client.room_id)

# Notice no decorator?
@client.on("comment")
async def on_comment(event: CommentEvent):
    #print(f"{event.user.nickname} -> {event.comment}")
    await commentQueue.put(f"{event.user.nickname} -> {event.comment}\n")
    

# Define handling an event via "callback"
#client.add_listener("comment", on_comment)
    
async def main():
    await asyncio.gather(client.start(), printQueue())
    #await client.start()
    #await printQueue()

async def printQueue():
    while True:
        if not commentQueue.empty():
            item = await commentQueue.get()
            print("Comment: {}".format(item))
        else:
            print("Waiting for new Comments...\n")
        
if __name__ == '__main__':
    asyncio.run(main())

运行输出 #

Waiting for new Comments...

Waiting for new Comments...

Waiting for new Comments...

Waiting for new Comments...

Connected to Room ID: 7181046628336356139
Comment: swift_sunsets -> 1

Comment: drosdye -> 1

Comment: what talent does charli have -> 1 2 3 4

Comment: what talent does charli have -> 1 2 3 4

Comment: I’m_paradox -> 1

Comment: I’m_paradox -> 1

Comment: Amir Morris -> 2

Waiting for new Comments...

Waiting for new Comments...

Waiting for new Comments...

问题出在printQueue函数的轮询逻辑上。当前代码会在队列空时持续高频输出等待信息，抢占了事件循环的资源，导致TikTokLive的on_comment回调无法及时执行，进而无法向队列中添加新评论。

asyncio.Queue.get()本身就是异步阻塞方法，会自动挂起直到队列中有元素，无需手动判断队列是否为空。修改printQueue函数如下：

async def printQueue():
    while True:
        item = await commentQueue.get()
        print("Comment: {}".format(item))
        # 标记任务完成，避免队列内部未完成计数累积（可选但推荐）
        commentQueue.task_done()

修改后，当队列空时，get()会让出事件循环资源，让TikTokLive的回调正常处理新评论并添加到队列中，程序就能持续接收并打印评论。

完整修改后的代码 #

import asyncio
from asyncio import Queue
from TikTokLive import TikTokLiveClient
from TikTokLive.types.events import CommentEvent, ConnectEvent

# Instantiate the client with the user's username
client: TikTokLiveClient = TikTokLiveClient(unique_id="@territorialwarfr")

# Comment Holders
commentQueue = asyncio.Queue()

@client.on("connect")
async def on_connect(_: ConnectEvent):
    print("Connected to Room ID:", client.room_id)

@client.on("comment")
async def on_comment(event: CommentEvent):
    await commentQueue.put(f"{event.user.nickname} -> {event.comment}\n")
    
async def main():
    await asyncio.gather(client.start(), printQueue())

async def printQueue():
    while True:
        item = await commentQueue.get()
        print("Comment: {}".format(item))
        commentQueue.task_done()

if __name__ == '__main__':
    asyncio.run(main())

内容的提问来源于stack exchange，提问作者kushy