需求：做一个程序等待用户输入，3秒内输入则会 echo 这个输入并立即退出。3秒内没输入则自动退出。

实现方法：

1. 线程（错误示范）

import time
import threading

def get_input():
    print('您输入了:', input('请在 3 秒内输入，否则结束程序。\n'))

wait_thread = threading.Thread(target=time.sleep, args=(3,))
input_thread = threading.Thread(target=get_input)
input_thread.daemon = True

wait_thread.start()
input_thread.start()

首先启动两个线程，并把等待输入的 get_input 设置成 daemon。于是 3 秒后，整个程序退出，3 秒内可以获得输入。
这个程序最大的问题在于，获得输入后并不会立即退出，而是无论如何都要等 3 秒。

2. 线程 + 队列

import threading
import queue

def get_input(q):
    print('您输入了:', input('请在 3 秒内输入，否则结束程序。\n'))
    q.put(1)

def wait(q):
    try:
        q.get(timeout=3)
    except queue.Empty:
        pass

q = queue.Queue()

wait_thread = threading.Thread(target=wait, args=(q,))
input_thread = threading.Thread(target=get_input, args=(q,))

input_thread.daemon = True

wait_thread.start()
input_thread.start()

这个程序完全实现了需求。原理在于给 q 设置超时退出。想要让 wait 提前退出，则往队列放东西。

3. 线程 + 事件

import threading

def get_input(e):
    print('您输入了:', input('请在 3 秒内输入，否则结束程序。\n'))
    e.set()

def wait_event(e):
    e.wait(timeout=3)

e = threading.Event()

wait_thread = threading.Thread(target=wait_event, args=(e,))
input_thread = threading.Thread(target=get_input, args=(e,))

input_thread.daemon = True

wait_thread.start()
input_thread.start()

这个版本基本我能想到的线程版本的最优版本了。代码简洁且只需要一个库。
Event() 对象可以用 wait 方法来阻塞。e.set() 的作用是解除阻塞。

4. 协程

import functools
import asyncio
import sys

async def timeout(loop):
    print('请在 3 秒内输入，否则结束程序。')
    await asyncio.sleep(3)
    loop.stop()

def echo(loop):
    print("您输入了: " + sys.stdin.readline(), end='')
    loop.stop()

loop = asyncio.get_event_loop()
asyncio.ensure_future(timeout(loop))
loop.add_reader(sys.stdin, functools.partial(echo, loop=loop))
loop.run_forever()

ensure_future 表示把协程注册到 event loop 里。
add_reader 表示把 sys.stdin 这个文件的输入事件和函数 echo 绑定，由于 echo 需要传参，因此用到了 functools.partial。
loop.run_forever 不需要参数，因为前面两行已经注册事件到 loop 了，就不用传参了。
一旦两个事件中任意一个完成，loop.close() 就触发。

延伸（错误示范）

import functools
import asyncio
import sys

async def timeout(loop):
    print('请在 3 秒内输入，否则结束程序。')
    await asyncio.sleep(3)

def echo(loop):
    print("您输入了: " + sys.stdin.readline(), end='')
    loop.stop()

loop = asyncio.get_event_loop()
future = asyncio.ensure_future(timeout(loop))
loop.add_reader(sys.stdin, functools.partial(echo, loop=loop))
loop.run_until_complete(future)

这里，run_until_complete 必须传入一个 Future 对象作为参数。于是我把 timeout 传入。
然而，这个程序有个问题：
在 echo 中触发 loop.stop 会报错，说 Future 没有完成就停止 loop 了。除非写成：

try:
    loop.run_until_complete(future)
except RuntimeError:
    pass

但是这样不够优雅，建议用 run_forever。

你还能有什么更好的实现方法的话，请留下评论，不吝赐教。

---

以上。