Python非阻塞input解决input()阻塞定时API请求问题

我开发了一款应用，需要每4秒向自有API发送GET请求以维持授权状态。当前遇到的问题是：代码中大量使用的input()函数会阻塞线程，导致授权请求无法正常发送。我曾尝试编写独立模块解决该阻塞问题，但阻塞问题依然存在；我花费了大量时间尝试实现非阻塞输入功能，几乎尝试了Stack Overflow上提及的所有相关方案。以下是我编写的非阻塞输入实现类（其中包含一个设计目标为与Python内置input()功能几乎一致的nbput()函数），请帮忙排查代码仍存在阻塞问题的原因，给出可行的修正方案：

原问题代码：

from pynput import keyboard
from pynput.keyboard import Key
import threading
import sys
from functools import partial

uinput = ''
lastKey = ''
modifierKey = ''


class NonBlockingInput:
    arg0 = ''

    def __init__(self):
        global listener
        listener = keyboard.Listener(on_press=self.on_press, on_release=self.on_release)
        print('1')

    def on_press(self, key):
        global lastKey
        global modifierKey
        try:
            sys.stdout.write(key.char)
            lastKey = key.char
        except AttributeError:
            if key == Key.space:
                lastKey = ' '
                sys.stdout.write(' ')
                return
            modifierKey = key

    def on_release(self, key):
        pass

    def nbinput(self, prompt):
        global uinput
        global listener
        global lastKey
        global modifierKey
        global arg0
        listener.start()
        sys.stdout.write(prompt)
        while True:
            if modifierKey == Key.enter:
                sys.stdout.write('\n')
                value_returned = partial(self.retrieved_data_func, arg0)
                break
            elif modifierKey == Key.backspace:
                spaceString = ''
                for _ in range(0, len(uinput)):
                    spaceString += ' '
                uinput = uinput[:-1]
                sys.stdout.write('\r')
                sys.stdout.write(spaceString)
                sys.stdout.write('\r')
                sys.stdout.write(uinput)
                modifierKey = ''
            else:
                uinput += lastKey
                lastKey = ''

    def retrieved_data_func(self):
        arg0 = 0
        return arg0


def nbput(prompt=''):
    global collectionThread
    nonBlockingInput = NonBlockingInput()
    collectionThread = threading.Thread(nonBlockingInput.nbinput(prompt))
    collectionThread.start()
    return NonBlockingInput.retrieved_data_func()


if __name__ == '__main__':
    print(nbput())

• 线程启动逻辑完全错误：threading.Thread(nonBlockingInput.nbinput(prompt)) 会先在主线程同步执行nbinput(prompt)，再将函数返回值（None）作为线程任务传入，本质上输入逻辑还是跑在主线程，死循环直接卡住主线程，和原生input()的阻塞效果没有区别。
• 全局变量滥用导致状态不可控：所有输入状态、监听器实例都用全局变量存储，多次调用会出现状态覆盖、监听器重复启动报错的问题，且跨线程修改变量没有加锁，会出现随机的输入异常。
• 输入轮询逻辑缺陷：死循环没有任何休眠，会跑满单个CPU核心；普通字符追加逻辑不管lastKey是否为空都执行拼接，会生成大量无效空字符；退格、回车等特殊键没有处理长按重复触发的问题。
• 返回值逻辑失效：nbput()启动线程后立刻调用返回函数，拿不到用户实际输入的内容，固定返回0。

修正点：

• 正确传递线程target与参数，确保输入逻辑运行在子线程
• 用类实例变量替代全局变量，封装所有输入状态
• 增加线程事件做输入完成通知，调用nbput()时可以同步等待输入结果，也可以根据需求改成异步回调
• 轮询增加短休眠降低CPU占用，修正按键拼接、特殊键处理逻辑
• 增加心跳线程示例，验证输入时心跳请求不会被阻塞

from pynput import keyboard
from pynput.keyboard import Key
import threading
import sys
import time


class NonBlockingInput:
    def __init__(self):
        # 用实例变量替代全局变量，避免多实例状态冲突
        self.uinput = ''
        self.lastKey = ''
        self.pressed_special = None
        self.input_done = threading.Event()
        self.listener = None
        self.result = None

    def on_press(self, key):
        try:
            # 普通可打印字符
            sys.stdout.write(key.char)
            sys.stdout.flush()
            self.lastKey = key.char
        except AttributeError:
            # 特殊按键处理
            if key == Key.space:
                sys.stdout.write(' ')
                sys.stdout.flush()
                self.lastKey = ' '
            elif key == Key.enter:
                self.pressed_special = Key.enter
            elif key == Key.backspace:
                self.pressed_special = Key.backspace

    def on_release(self, key):
        pass

    def _nbinput_runner(self, prompt):
        sys.stdout.write(prompt)
        sys.stdout.flush()
        self.listener = keyboard.Listener(on_press=self.on_press, on_release=self.on_release)
        self.listener.start()

        while not self.input_done.is_set():
            # 处理回车：输入完成
            if self.pressed_special == Key.enter:
                sys.stdout.write('\n')
                sys.stdout.flush()
                self.result = self.uinput
                self.input_done.set()
                self.listener.stop()
                break
            # 处理退格
            elif self.pressed_special == Key.backspace:
                if len(self.uinput) > 0:
                    self.uinput = self.uinput[:-1]
                    # 光标回退、覆盖最后一个字符、再回退
                    sys.stdout.write('\b \b')
                    sys.stdout.flush()
                self.pressed_special = None
            # 追加普通字符
            else:
                if self.lastKey:
                    self.uinput += self.lastKey
                    self.lastKey = ''
            # 加短休眠，避免占满CPU
            time.sleep(0.01)


def nbput(prompt=''):
    nbi = NonBlockingInput()
    # 正确传线程target和参数，不要直接调用函数
    input_thread = threading.Thread(target=nbi._nbinput_runner, args=(prompt,), daemon=True)
    input_thread.start()
    # 等待输入完成返回结果，如果需要完全异步可以去掉wait，改成回调逻辑
    nbi.input_done.wait()
    return nbi.result


# 测试：模拟4秒一次的授权心跳
def auth_heartbeat():
    while True:
        print(f"\n[心跳] 发送授权保活请求，时间：{time.strftime('%H:%M:%S')}")
        time.sleep(4)


if __name__ == '__main__':
    # 启动心跳守护线程
    threading.Thread(target=auth_heartbeat, daemon=True).start()
    time.sleep(0.5)
    while True:
        user_input = nbput("请输入内容：")
        print(f"收到输入：{user_input}")

注：Windows环境运行pynput需要管理员权限，macOS需要给终端开启辅助功能权限，否则键盘监听会失效。如果不想用pynput这类需要权限的库，Windows平台可以用msvcrt模块实现非阻塞输入，Linux/macOS可以通过修改tty属性实现，不需要额外依赖。

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