Python之asyncio模块的使用

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

# 开头定义async,表示要在协程运行，不定义的话，循环监听增加不了
async def coroutine():
    print(‘协程运行...‘)

# 定义一个事件循环监听
event_loop = asyncio.get_event_loop()

try:
    print(‘协程开始...‘)
    coroutine_obj = coroutine()
    print(‘进入事件循环监听...‘)
    event_loop.run_until_complete(coroutine())  # run_until_complete翻译成中文：一直运行到完成为止
finally:
    print(‘关闭事件循环监听..‘)
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

# 开头定义async,表示要在协程运行，不定义的话，循环监听增加不了
async def coroutine():
    print(‘协程运行...‘)
    return ‘ok‘

# 定义一个事件循环监听
event_loop = asyncio.get_event_loop()

try:
    coroutine_obj = coroutine()
    return_value = event_loop.run_until_complete(coroutine())  # run_until_complete翻译成中文：一直运行到完成为止
    print(‘coroutine()返回值：‘, return_value)
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

# 开头定义async,表示要在协程运行，不定义的话，循环监听增加不了
async def coroutine():
    print(‘coroutine内部运行‘)

    print(‘等待task_1运行结果‘)
    task1 = await task_1() #await作用：控制运行流程，按顺序执行，即等待该函数运行完成，再继续往后执行

    print(‘等待task_2运行结果‘)
    task2 = await task_2(task1)

    return (task1, task2)

async def task_1():
    print(‘task_1内部运行‘)
    return ‘task_1 ok‘


async def task_2(arg):
    print(‘task_2内部运行‘)
    return ‘task_2 arg:{}‘.format(arg)

# 定义一个事件循环监听
event_loop = asyncio.get_event_loop()

try:
    coroutine_obj = coroutine()
    return_value = event_loop.run_until_complete(coroutine())  # run_until_complete翻译成中文：一直运行到完成为止
    print(‘coroutine()返回值：‘, return_value)
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

# 开头定义async,表示要在协程运行，不定义的话，循环监听增加不了
@asyncio.coroutine
def coroutine():
    print(‘coroutine内部运行‘)

    print(‘等待task_1运行结果‘)
    task1 = yield from task_1() #await作用：控制运行流程，按顺序执行，即等待该函数运行完成，再继续往后执行

    print(‘等待task_2运行结果‘)
    task2 = yield from task_2(task1)

    return (task1, task2)

@asyncio.coroutine
async def task_1():
    print(‘task_1内部运行‘)
    return ‘task_1 ok‘

@asyncio.coroutine
async def task_2(arg):
    print(‘task_2内部运行‘)
    return ‘task_2 arg:{}‘.format(arg)

# 定义一个事件循环监听
event_loop = asyncio.get_event_loop()

try:
    coroutine_obj = coroutine()
    return_value = event_loop.run_until_complete(coroutine())  # run_until_complete翻译成中文：一直运行到完成为止
    print(‘coroutine()返回值：‘, return_value)
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import functools

def callback(arg, *, kwarg=‘default‘):
    print(‘回调函数arg={},kwarg={}‘.format(arg, kwarg))

async def main(loop):
    print(‘注册回调函数‘)
    loop.call_soon(callback, 1)  # 执行回调函数，传入参数1
    wrapped = functools.partial(callback, kwarg=‘not default‘)  # 利用偏函数，给kwarg传默认值
    loop.call_soon(wrapped, 2)  # 执行回调函数，传入参数2
    await asyncio.sleep(0.5)

event_loop = asyncio.get_event_loop()
try:
    print(‘进入事件循环监听‘)
    event_loop.run_until_complete(main(event_loop))  # 将事件循环对象传入main函数中
finally:
    print(‘关闭事件循环监听‘)
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

def callback(arg):
    print(‘回调函数arg={}‘.format(arg))

async def main(loop):
    print(‘注册回调函数‘)
    loop.call_later(1,callback,‘延时1秒回调参数1‘)
    loop.call_later(1,callback,‘延时1秒回调参数2‘)
    loop.call_soon(callback,‘讯速的回调参数‘)
    await asyncio.sleep(3)

event_loop = asyncio.get_event_loop()
try:
    print(‘进入事件循环监听‘)
    event_loop.run_until_complete(main(event_loop))  # 将事件循环对象传入main函数中
finally:
    print(‘关闭事件循环监听‘)
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import time

def callback(arg, loop):
    print(‘回调函数arg={} 回调的时间time={}‘.format(arg, loop.time()))

async def main(loop):
    now = loop.time()
    print(‘时钟时间:{}‘.format(time.time()))
    print(‘时事件循环时间:{}‘.format(loop.time()))
    print(‘注册回调函数‘)
    loop.call_at(now + 1, callback, ‘参数1‘, loop)
    loop.call_at(now + 2, callback, ‘参数2‘, loop)
    loop.call_soon(callback, ‘讯速的回调参数‘, loop)
    await asyncio.sleep(4)

event_loop = asyncio.get_event_loop()
try:
    print(‘进入事件循环监听‘)
    event_loop.run_until_complete(main(event_loop))  # 将事件循环对象传入main函数中
finally:
    print(‘关闭事件循环监听‘)
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

def mark_done(future, result):
    """标记完成的函数"""
    print(‘设置 Future 返回结果 {}‘.format(result))
    future.set_result(result)

event_loop = asyncio.get_event_loop()
try:
    all_done = asyncio.Future()
    print(‘调度标记完成的函数‘)
    event_loop.call_soon(mark_done, all_done, ‘这个是调度传入的数据‘)
    result = event_loop.run_until_complete(all_done)
    print(‘运行返回的结果:{}‘.format(result))
finally:
    print(‘关闭事件循环监听‘)
    event_loop.close()

print(‘Future 返回的结果: {}‘.format(all_done.result()))
"""
结论：
    返回结果可以从两个地方获取：
    1、result = event_loop.run_until_complete(all_done)
    2、Future.result()
"""

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

def mark_done(future, result):
    """标记完成的函数"""
    print(‘设置 Future 返回结果 {}‘.format(result))
    future.set_result(result)

async def main(loop):
    all_done = asyncio.Future()
    print(‘调度标记完成的函数‘)
    loop.call_soon(mark_done, all_done, ‘这个是调度传入的数据‘)
    result = await all_done  # await作用：等all_done返回结果，再往下运行
    print(‘mark_done()执行完成，返回值 : {}‘.format(result))

event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(main(event_loop))
finally:
    print(‘关闭事件循环监听‘)
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import functools

def callback(future, n):
    print(‘{}: future 完成: {}‘.format(n, future.result()))


async def register_callbacks(future_obj):
    print(‘将回调函数注册到Future中‘)

    # 这里需要注意的是add_done_callback函数，还为把当前实例对象作为参数，传给函数，所以回调函数多一个callback(future, n)
    future_obj.add_done_callback(functools.partial(callback, n=1))
    future_obj.add_done_callback(functools.partial(callback, n=2))

async def main(future_obj):
    # 注册future的回调函数
    await register_callbacks(future_obj)
    print(‘设置Future返回结果‘)
    future_obj.set_result(‘the result‘)

event_loop = asyncio.get_event_loop()
try:
    # 创建一个future实例
    future_obj = asyncio.Future()

    # 增future实例传给main函数处理
    event_loop.run_until_complete(main(future_obj))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def task_func():
    print(‘task_func 执行完成‘)
    return ‘task_func返回值ok‘

async def main(loop):
    print(‘创建任务‘)
    task = loop.create_task(task_func())
    print(‘等待task的结果 {}‘.format(task))
    return_value = await task #直到遇到await，上面的task开始运行
    print(‘已完成任务{}‘.format(task)) #经过上面的运行，task里面已经有result执行结果
    print(‘return value: {}‘.format(return_value))

event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(main(event_loop))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def task_func():
    print(‘task_func 执行完成‘)
    return ‘task_func返回值ok‘

async def main(loop):
    print(‘创建任务‘)
    task = loop.create_task(task_func())
    print(‘取消任务‘)
    task.cancel()
    print(‘取消任务结果 {}‘.format(task))
    try:
        await task #直到遇到await，上面的task开始运行
    except asyncio.CancelledError:
        print(‘从已取消的任务中捕获错误‘)
    else:
        print(‘任务执行结果 {}‘.format(task))

event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(main(event_loop))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def task_func():
    print(‘task_func睡眠‘)
    try:
        await asyncio.sleep(1)
    except asyncio.CancelledError:
        print(‘task_func 任务取消‘)
        raise
    return ‘task_func返回值ok‘

def task_canceller(task_obj):
    print(‘task_canceller运行‘)
    task_obj.cancel()
    print(‘取消task_obj任务‘)

async def main(loop):
    print(‘创建任务‘)
    task = loop.create_task(task_func())
    loop.call_soon(task_canceller, task)
    try:
        await task  # 直到遇到await，上面的task开始运行
    except asyncio.CancelledError:
        print(‘从已取消的任务中捕获错误‘)
    else:
        print(‘任务执行结果 {}‘.format(task))

event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(main(event_loop))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def wrapped():
    print(‘wrapped 运行‘)
    return ‘wrapped result‘

async def inner(task):
    print(‘inner: 开始运行‘)
    print(‘inner: task {!r}‘.format(task))
    result = await task
    print(‘inner: task 返回值 {!r}‘.format(result))

async def start_task():
    print(‘开始创建task‘)
    task = asyncio.ensure_future(wrapped())
    print(‘等待inner运行‘)
    await inner(task)
    print(‘starter: inner returned‘)

event_loop = asyncio.get_event_loop()
try:
    print(‘进程事件循环‘)
    result = event_loop.run_until_complete(start_task())
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def phase(i):
    print(‘phase形参传入值{}‘.format(i))
    await asyncio.sleep(0.1 * i)
    print(‘完成phase的次数{}‘.format(i))
    return ‘phase {} result‘.format(i)

async def main(num_phase):
    print(‘main函数开始‘)
    phases = [
        phase(i) for i in range(num_phase)
    ]
    print(‘等待phases里面的多个函数执行完成‘)
    completed, pending = await asyncio.wait(phases)  # completed : 运行完成存在这里 ，pending ： 没有运行完成存在这里
    for_completed_results = [t.result() for t in completed]
    print(‘for_completed_results : {}‘.format(for_completed_results))

event_loop = asyncio.get_event_loop()
try:
    print(‘进程事件循环‘)
    result = event_loop.run_until_complete(main(3))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def phase(i):
    print(‘phase形参传入值{}‘.format(i))
    try:
        await asyncio.sleep(0.1 * i)
    except asyncio.CancelledError:
        print(‘phase {} 取消‘.format(i))
    else:
        print(‘完成phase的次数{}‘.format(i))
        return ‘phase {} result‘.format(i)

async def main(num_phase):
    print(‘main函数开始‘)
    phases = [
        phase(i) for i in range(num_phase)
    ]
    print(‘等待phases里面的多个函数执行完成‘)
    completed, pending = await asyncio.wait(phases, timeout=0.1)  # completed : 运行完成存在这里 ，pending ： 没有运行完成存在这里

    print(‘completed长度：{}，pending长度 :{}‘.format(len(completed), len(pending)))

    if pending:
        print(‘取消未完成的任务‘)
        for t in pending:
            t.cancel()
    print(‘main函数执行完成‘)


event_loop = asyncio.get_event_loop()
try:
    print(‘进程事件循环‘)
    result = event_loop.run_until_complete(main(3))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def phase1():
    print(‘phase1运行中‘)
    await asyncio.sleep(2)
    print(‘phase1运行完成‘)
    return ‘phase1 result‘

async def phase2():
    print(‘phase2运行中‘)
    await asyncio.sleep(1)
    print(‘phase2运行完成‘)
    return ‘phase2 result‘

async def main():
    print(‘main函数开始‘)
    results = await asyncio.gather(
        phase1(),
        phase2()
    )
    print(‘results : {}‘.format(results))

event_loop = asyncio.get_event_loop()
try:
    print(‘进程事件循环‘)
    result = event_loop.run_until_complete(main())
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def phase(i):
    print(‘phase {} 运行中‘.format(i))
    await asyncio.sleep(0.5 - (0.1 * i))
    print(‘phase {} 运行完成‘.format(i))
    return ‘phase {} result‘.format(i)

async def main(num_phases):
    print(‘main函数开始‘)
    phases = [
        phase(i) for i in range(num_phases)
    ]
    print(‘等待phases运行完成‘)
    results = []
    for next_to_complete in asyncio.as_completed(phases):
        task_result = await next_to_complete
        print(‘接到到task_result : {}‘.format(task_result))
        results.append(task_result)
    print(‘results : {}‘.format(results))
    return results

event_loop = asyncio.get_event_loop()
try:
    print(‘进程事件循环‘)
    event_loop.run_until_complete(main(3))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import functools

def unlock(lock):
    print(‘回调释放锁‘)
    lock.release()

async def coro1(lock):
    """with方式获得锁"""
    async with lock:
        print(‘coro1 打开锁运算‘)
    print(‘coro1 锁已释放‘)

async def coro2(lock):
    """传统方式获取锁"""
    await lock.acquire()
    try:
        print(‘coro2 打开锁运算‘)
    finally:
        print(‘coro2 锁已释放‘)
        lock.release()

async def main(loop):
    lock = asyncio.Lock()
    print(‘启动协程之前获取锁‘)
    await lock.acquire()
    print(‘获得锁 {}‘.format(lock.locked()))

    # 运行完成，回调解锁
    loop.call_later(0.1, functools.partial(unlock, lock))

    print(‘等待协程运行‘)
    await asyncio.wait([coro1(lock), coro2(lock)])


event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(main(event_loop))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import functools

def set_event(event):
    print(‘回调设置event‘)
    event.set()

async def coro1(event):
    print(‘coro1 等待事件‘)
    await event.wait()
    print(‘coro1 触发运行‘)

async def coro2(event):
    print(‘coro2 等待事件‘)
    await event.wait()
    print(‘coro2 触发运行‘)

async def main(loop):
    event = asyncio.Event()
    print(‘event开始之前状态:{}‘.format(event.is_set()))
    loop.call_later(0.1, functools.partial(set_event, event))  # 延时0.1秒后，回调set_event函数
    await asyncio.wait([coro1(event), coro2(event)])
    print(‘event开始之后状态:{}‘.format(event.is_set()))

event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(main(event_loop))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def consumer(condition_obj, i):
    async with condition_obj:
        print(‘消费者 {} 等待中‘.format(i))
        await condition_obj.wait()
        print(‘消费者 {} 触发‘.format(i))
    print(‘消费者 {} 消费结束‘.format(i))

async def manipulate_condition(condition_obj):
    print(‘开始操作condition‘)
    await asyncio.sleep(0.1)
    for i in range(3):
        async with condition_obj:
            print(‘通知消费者 {}‘.format(i))
            condition_obj.notify(i)
        await asyncio.sleep(0.1)

    async with condition_obj:
        print(‘通知其它所有的消费者‘)
        condition_obj.notify_all()
    print(‘操作condition结束‘)

async def main(loop):
    # 创建一个操作状态的对象
    condition_obj = asyncio.Condition()

    # 运5个消费者函数
    consumers = [
        consumer(condition_obj, i) for i in range(5)
    ]

    # 创建一个操作状态的任务
    loop.create_task(manipulate_condition(condition_obj))

    # 等待consumers所有的函数执行完成
    await asyncio.wait(consumers)

event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(main(event_loop))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def consumer(n, q):
    print(‘消费者 {} 开始‘.format(n))
    while True:
        print(‘消费费 {} 等待消费‘.format(n))
        item = await q.get()
        print(‘消费者 {} 消费了 {}‘.format(n, item))
        if item is None:
            q.task_done()
            break
        else:
            await asyncio.sleep(0.01 * item)
            q.task_done()
    print(‘消费者 {} 结束‘.format(n))

async def producer(q, num_worker):
    print(‘生产者 开始‘)

    for i in range(num_worker * 3):
        await q.put(i)
        print(‘生产者 增加数据 {} 到队列中‘.format(i))

    print(‘生产者 增加停止信号到队列中‘)
    for i in range(num_worker):
        await q.put(None)
    print(‘生产者 等待队列清空‘)
    await q.join()
    print(‘生产者 结束‘)

async def main(loop, num_consumers):
    # 创建一个队列，最大的长度是num_consumers
    q = asyncio.Queue(maxsize=num_consumers)

    consumers = [
        loop.create_task(consumer(i, q)) for i in range(num_consumers)
    ]

    producer_task = loop.create_task(producer(q, num_consumers))

    await asyncio.wait(consumers + [producer_task])

event_loop = asyncio.get_event_loop()
try:
    event_loop.run_until_complete(main(event_loop, 2))
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import logging
import sys

SERVER_ADDRESS = [‘localhost‘, 8000]

class EchoServer(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.address = transport.get_extra_info(‘peername‘)
        self.log = logging.getLogger(
            ‘EchoServer_{}_{}‘.format(*self.address)
        )
        self.log.debug(‘接收连接‘)

    def data_received(self, data):
        self.log.debug(‘接收数据 {}‘.format(data))
        self.transport.write(data)
        self.log.debug(‘发送数据 {}‘.format(data))

    def eof_received(self):
        self.log.debug(‘接收数据 EOF‘)
        if self.transport.can_write_eof():
            self.transport.write_eof()

    def connection_lost(self, exc):
        if exc:
            self.log.error(‘错误 {}‘.format(exc))
        else:
            self.log.debug(‘服务关闭‘)
        super(EchoServer, self).connection_lost(exc)

logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(name)s : %(message)s‘,
    stream=sys.stderr
)

log = logging.getLogger(‘main‘)

event_loop = asyncio.get_event_loop()

factory = event_loop.create_server(EchoServer, *SERVER_ADDRESS)

server = event_loop.run_until_complete(factory)

log.debug(‘开始运行 IP:{} Port:{}‘.format(*SERVER_ADDRESS))

try:
    event_loop.run_forever()
finally:
    log.debug(‘关闭服务‘)
    server.close()
    event_loop.run_until_complete(server.wait_closed())
    log.debug(‘关闭事件循环‘)
    event_loop.close()

#!/usr/bin/env python3
# encoding: utf-8

import asyncio
import functools
import logging
import sys

MESSAGES = [
    b‘This is the message. ‘,
    b‘It will be sent ‘,
    b‘in parts.‘,
]
SERVER_ADDRESS = (‘localhost‘, 8000)

class EchoClient(asyncio.Protocol):

    def __init__(self, messages, future):
        super().__init__()
        self.messages = messages
        self.log = logging.getLogger(‘EchoClient‘)
        self.f = future

    def connection_made(self, transport):
        self.transport = transport
        self.address = transport.get_extra_info(‘peername‘)
        self.log.debug(
            ‘连接服务器IP：{} port ：{}‘.format(*self.address)
        )

        for msg in self.messages:
            transport.write(msg)
            self.log.debug(‘发送数据 {!r}‘.format(msg))
        if transport.can_write_eof():
            transport.write_eof()

    def data_received(self, data):
        self.log.debug(‘received {!r}‘.format(data))

    def eof_received(self):
        self.log.debug(‘接收到 EOF‘)
        self.transport.close()
        if not self.f.done():
            self.f.set_result(True)

    def connection_lost(self, exc):
        self.log.debug(‘服务器关闭连接‘)
        self.transport.close()
        if not self.f.done():
            self.f.set_result(True)
        super().connection_lost(exc)

#设置日志级别
logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(name)s: %(message)s‘,
    stream=sys.stderr,
)

#打印日志标题
log = logging.getLogger(‘main‘)

#创建一个事件循环
event_loop = asyncio.get_event_loop()

#创建客户端的Future
client_completed = asyncio.Future()

#利用偏函数自动传参给EchoClient实例化类
client_factory = functools.partial(
    EchoClient,
    messages=MESSAGES,
    future=client_completed,
)

#创建一个事件循环连接
factory_coroutine = event_loop.create_connection(
    client_factory,
    *SERVER_ADDRESS,
)

log.debug(‘等待客户端运行完成‘)
try:
    event_loop.run_until_complete(factory_coroutine)
    event_loop.run_until_complete(client_completed)
finally:
    log.debug(‘关闭事件循环‘)
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import logging
import sys

SERVER_ADDRESS = (‘localhost‘, 8000)

async def echo(reader, writer):
    address = writer.get_extra_info(‘peername‘)
    log = logging.getLogger(‘echo_{}_{}‘.format(*address))
    log.debug(‘开始接受连接‘)
    while True:
        data = await reader.read(128)
        if data:
            log.debug(‘接受的数据 : {}‘.format(data))
            writer.write(data)
            await writer.drain()
            log.debug(‘发送数据:{}‘.format(data))
        else:
            log.debug(‘关闭连接‘)
            writer.close()
            return

logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(name)s : %(message)s‘,
    stream=sys.stderr
)

log = logging.getLogger(‘main‘)

event_loop = asyncio.get_event_loop()

factory = asyncio.start_server(echo, *SERVER_ADDRESS)
server = event_loop.run_until_complete(factory)
log.debug(‘开始启动服务 IP:{},Port:{}‘.format(*SERVER_ADDRESS))

try:
    event_loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    log.debug(‘关闭服务端‘)
    server.close()
    event_loop.run_until_complete(server.wait_closed())
    log.debug(‘关闭事件循环‘)
    event_loop.close()

#!/usr/bin/env python3
# encoding: utf-8

import asyncio
import logging
import sys

MESSAGES = [
    b‘This is the message. ‘,
    b‘It will be sent ‘,
    b‘in parts.‘,
]
SERVER_ADDRESS = (‘localhost‘, 8000)

async def echo_client(address, messages):
    log = logging.getLogger(‘echo_client‘)

    log.debug(‘连接服务器 to {} port {}‘.format(*address))

    # 创建与服务端连接
    reader, writer = await asyncio.open_connection(*address)

    for msg in messages:
        writer.write(msg)
        log.debug(‘发送数据： {}‘.format(msg))

    # 判断是否发送结束标记
    if writer.can_write_eof():
        writer.write_eof()
    # 等待所有发送完成
    await writer.drain()

    log.debug(‘等待服务器响应‘)
    while True:
        data = await reader.read(128)
        if data:
            log.debug(‘接收服务器数据 ：{}‘.format(data))
        else:
            log.debug(‘关闭与服务器的连接‘)
            writer.close()
            return

logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(name)s: %(message)s‘,
    stream=sys.stderr,
)
log = logging.getLogger(‘main‘)

event_loop = asyncio.get_event_loop()

try:
    event_loop.run_until_complete(
        echo_client(SERVER_ADDRESS, MESSAGES)
    )
finally:
    log.debug(‘closing event loop‘)
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import logging
import sys
import ssl

SERVER_ADDRESS = (‘localhost‘, 8000)

async def echo(reader, writer):
    address = writer.get_extra_info(‘peername‘)
    log = logging.getLogger(‘echo_{}_{}‘.format(*address))
    log.debug(‘开始接受连接‘)
    while True:
        data = await reader.read(128)

        #因为ssl不支持EOF结束，所以需要用‘\x00‘结束
        terminate = data.endswith(b‘\x00‘)
        data = data.rstrip(b‘\x00‘)
        if data:
            log.debug(‘接受的数据 : {}‘.format(data))
            writer.write(data)
            await writer.drain()
            log.debug(‘发送数据:{}‘.format(data))
        if not data or terminate:
            log.debug(‘关闭连接‘)
            writer.close()
            return

logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(name)s : %(message)s‘,
    stream=sys.stderr
)

log = logging.getLogger(‘main‘)

event_loop = asyncio.get_event_loop()

# 创建SSL所需要的对象
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.check_hostname = False
ssl_context.load_cert_chain(‘test_ssl.crt‘, ‘test_ssl.key‘)

factory = asyncio.start_server(echo, *SERVER_ADDRESS, ssl=ssl_context)
server = event_loop.run_until_complete(factory)
log.debug(‘开始启动服务 IP:{},Port:{}‘.format(*SERVER_ADDRESS))

try:
    event_loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    log.debug(‘关闭服务端‘)
    server.close()
    event_loop.run_until_complete(server.wait_closed())
    log.debug(‘关闭事件循环‘)
    event_loop.close()

#!/usr/bin/env python3
# encoding: utf-8

import asyncio
import logging
import sys
import ssl

MESSAGES = [
    b‘This is the message. ‘,
    b‘It will be sent ‘,
    b‘in parts.‘,
]
SERVER_ADDRESS = (‘localhost‘, 8000)

async def echo_client(address, messages):
    log = logging.getLogger(‘echo_client‘)

    # 客户端ssl所需要带证书访问
    ssl_context = ssl.create_default_context(
        ssl.Purpose.SERVER_AUTH
    )
    ssl_context.check_hostname = False
    ssl_context.load_verify_locations(‘test_ssl.crt‘)

    log.debug(‘连接服务器 to {} port {}‘.format(*address))

    # 创建与服务端连接
    reader, writer = await asyncio.open_connection(*address, ssl=ssl_context)

    for msg in messages:
        writer.write(msg)
        log.debug(‘发送数据： {}‘.format(msg))

    # 判断是否发送结束标记
    # 非ssl
    # if writer.can_write_eof():
    #     writer.write_eof()

    # ssl
    writer.write(b‘\x00‘)

    # 等待所有发送完成
    await writer.drain()

    log.debug(‘等待服务器响应‘)
    while True:
        data = await reader.read(128)
        if data:
            log.debug(‘接收服务器数据 ：{}‘.format(data))
        else:
            log.debug(‘关闭与服务器的连接‘)
            writer.close()
            return

logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(name)s: %(message)s‘,
    stream=sys.stderr,
)
log = logging.getLogger(‘main‘)

event_loop = asyncio.get_event_loop()

try:
    event_loop.run_until_complete(
        echo_client(SERVER_ADDRESS, MESSAGES)
    )
finally:
    log.debug(‘closing event loop‘)
    event_loop.close()

#!/usr/bin/env python3
# encoding: utf-8

#end_pymotw_header
import asyncio
import functools

class DFProtocol(asyncio.SubprocessProtocol):

    FD_NAMES = [‘stdin‘, ‘stdout‘, ‘stderr‘]

    def __init__(self, done_future):
        self.done = done_future
        self.buffer = bytearray()
        super().__init__()

    def connection_made(self, transport):
        print(‘process started {}‘.format(transport.get_pid()))
        self.transport = transport

    def pipe_data_received(self, fd, data):
        print(‘read {} bytes from {}‘.format(len(data),
                                             self.FD_NAMES[fd]))
        if fd == 1:
            self.buffer.extend(data)

    def process_exited(self):
        print(‘process exited‘)
        return_code = self.transport.get_returncode()
        print(‘return code {}‘.format(return_code))
        if not return_code:
            cmd_output = bytes(self.buffer).decode()
            results = self._parse_results(cmd_output)
        else:
            results = []
        self.done.set_result((return_code, results))

    def _parse_results(self, output):
        print(‘parsing results‘)
        if not output:
            return []
        lines = output.splitlines()
        headers = lines[0].split()
        devices = lines[1:]
        results = [
            dict(zip(headers, line.split()))
            for line in devices
        ]
        return results


async def run_df(loop):
    print(‘in run_df‘)

    cmd_done = asyncio.Future(loop=loop)
    factory = functools.partial(DFProtocol, cmd_done)
    proc = loop.subprocess_exec(
        factory,
        ‘df‘, ‘-hl‘,
        stdin=None,
        stderr=None,
    )
    try:
        print(‘launching process‘)
        transport, protocol = await proc
        print(‘waiting for process to complete‘)
        await cmd_done
    finally:
        transport.close()

    return cmd_done.result()


event_loop = asyncio.get_event_loop()
try:
    return_code, results = event_loop.run_until_complete(
        run_df(event_loop)
    )
finally:
    event_loop.close()

if return_code:
    print(‘error exit {}‘.format(return_code))
else:
    print(‘\nFree space:‘)
    for r in results:
        print(‘{Mounted:25}: {Avail}‘.format(**r))

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

def _parse_results(output):
    print(‘解析结果‘)
    if not output:
        return []
    lines = output.splitlines()
    headers = lines[0].split()
    devices = lines[1:]
    results = [
        dict(zip(headers, line.split())) for line in devices
    ]
    return results

async def run_df():
    print(‘run_df函数运行‘)
    buffer = bytearray()
    create = asyncio.create_subprocess_exec(
        ‘df‘, ‘-h‘,
        stdout=asyncio.subprocess.PIPE
    )

    print(‘df -h开始运行‘)
    proc = await create
    print(‘进程开始 {}‘.format(proc.pid))

    while True:
        line = await proc.stdout.readline()
        print(‘读取 : {}‘.format(line))
        if not line:
            print(‘命令不再输出‘)
            break
        buffer.extend(line)
    print(‘等待进程运行完成‘)
    await proc.wait()
    return_code = proc.returncode
    print(‘运行返回码:{}‘.format(return_code))
    if not return_code:
        cmd_output = bytes(buffer).decode()
        results = _parse_results(cmd_output)
    else:
        results = []

    return (return_code, results)

event_loop = asyncio.get_event_loop()
try:
    return_code, results = event_loop.run_until_complete(
        run_df()
    )
finally:
    event_loop.close()

if return_code:
    print(‘错误退出，错误信息：{}‘.format(return_code))
else:
    print(‘运行结果:‘)
    for r in results:
        print(‘{Mounted:25}:{Avail}‘.format(**r))

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio

async def to_upper(input):
    print(‘进程转大写的to_upper函数‘)

    create = asyncio.create_subprocess_exec(
        ‘tr‘, ‘[:lower:]‘, ‘[:upper:]‘,
        stdout=asyncio.subprocess.PIPE,
        stdin=asyncio.subprocess.PIPE,
    )
    print(‘等待子进程运行完成‘)
    proc = await create
    print(‘子进程PID {}‘.format(proc.pid))

    print(‘查看子进程运行的标准输出和错误‘)
    stdout, stderr = await proc.communicate(input.encode())

    print(‘等待子进程完成‘)
    await proc.wait()

    return_code = proc.returncode
    print(‘return code {}‘.format(return_code))
    if not return_code:
        results = bytes(stdout).decode()
    else:
        results = ‘‘
    return (return_code, results)

MESSAGE = """
This message will be converted
to all caps.
"""

event_loop = asyncio.get_event_loop()
try:
    return_code, results = event_loop.run_until_complete(
        to_upper(MESSAGE)
    )
finally:
    event_loop.close()

if return_code:
    print(‘错误时，退出的返回状态码 {}‘.format(return_code))
else:
    print(‘源数据: {!r}‘.format(MESSAGE))
    print(‘处理过的数据 : {!r}‘.format(results))

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import functools
import os
import signal

def signal_handler(name):
    print(‘正在处理信号 : {}‘.format(name))

event_loop = asyncio.get_event_loop()

# 给信号绑定处理的事件
event_loop.add_signal_handler(
    signal.SIGHUP,
    functools.partial(signal_handler, name=‘SIGHUP‘),
)
event_loop.add_signal_handler(
    signal.SIGUSR1,
    functools.partial(signal_handler, name=‘SIGUSR1‘),
)
event_loop.add_signal_handler(
    signal.SIGINT,
    functools.partial(signal_handler, name=‘SIGINT‘),
)

async def send_signals():
    pid = os.getpid()
    print(‘开始发送信号给PID：{}‘.format(pid))

    for name in [‘SIGHUP‘, ‘SIGHUP‘, ‘SIGUSR1‘, ‘SIGINT‘]:
        print(‘发送信号名字：{}‘.format(name))
        # 跟linux 命令kill一样，利用pid结束进程
        os.kill(pid, getattr(signal, name))
        print(‘放弃控制‘)
        await asyncio.sleep(0.01)
    return

try:
    event_loop.run_until_complete(send_signals())
finally:
    event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import logging
import sys
import concurrent.futures
import time

def blocks(n):
    log = logging.getLogger(‘blocks({})‘.format(n))
    log.info(‘运行‘)
    time.sleep(0.1)
    log.info(‘done‘)
    return n ** 2

async def run_blocking_tasks(executor):
    """运行阻塞的任务"""
    log = logging.getLogger(‘run_blocking_tasks‘)
    log.info(‘开始运行‘)
    log.info(‘创建执行任务‘)
    loop = asyncio.get_event_loop()
    blocking_tasks = [
        loop.run_in_executor(executor, blocks, i) for i in range(6)
    ]
    log.info(‘等待执行任务‘)
    completed, pending = await asyncio.wait(blocking_tasks)
    results = [t.result() for t in completed]
    log.info(‘运行结果: {!r}‘.format(results))

    log.info(‘exitrun_blocking_tasks 退出‘)

if __name__ == ‘__main__‘:
    logging.basicConfig(
        level=logging.INFO,
        format=‘%(threadName)10s %(name)18s: %(message)s‘,
        stream=sys.stderr
    )

    # 创建一个线程池执行器，最大开启3个工作线程
    executor = concurrent.futures.ThreadPoolExecutor(
        max_workers=3
    )

    # 创建一个事件循环
    event_loop = asyncio.get_event_loop()
    try:
        event_loop.run_until_complete(run_blocking_tasks(executor))
    finally:
        event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import logging
import sys
import concurrent.futures
import time

def blocks(n):
    log = logging.getLogger(‘blocks({})‘.format(n))
    log.info(‘运行‘)
    time.sleep(0.1)
    log.info(‘done‘)
    return n ** 2

async def run_blocking_tasks(executor):
    """运行阻塞的任务"""
    log = logging.getLogger(‘run_blocking_tasks‘)
    log.info(‘开始运行‘)
    log.info(‘创建执行任务‘)
    loop = asyncio.get_event_loop()
    blocking_tasks = [
        loop.run_in_executor(executor, blocks, i) for i in range(6)
    ]
    log.info(‘等待执行任务‘)
    completed, pending = await asyncio.wait(blocking_tasks)
    results = [t.result() for t in completed]
    log.info(‘运行结果: {!r}‘.format(results))

    log.info(‘exitrun_blocking_tasks 退出‘)

if __name__ == ‘__main__‘:
    logging.basicConfig(
        level=logging.INFO,
        format=‘PID %(process)5s %(name)18s: %(message)s‘,
        stream=sys.stderr
    )

    # 创建一个线程池执行器，最大开启3个工作线程
    executor = concurrent.futures.ProcessPoolExecutor(
        max_workers=3
    )

    # 创建一个事件循环
    event_loop = asyncio.get_event_loop()
    try:
        event_loop.run_until_complete(run_blocking_tasks(executor))
    finally:
        event_loop.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import argparse
import asyncio
import logging
import sys
import time
import warnings

#接收命令行
parser = argparse.ArgumentParser(‘debugging asyncio‘)
parser.add_argument(
    ‘-v‘,
    dest=‘verbose‘,
    default=False,
    action=‘store_true‘,
)
args = parser.parse_args()

#设置日志级别
logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(levelname)7s: %(message)s‘,
    stream=sys.stderr,
)
LOG = logging.getLogger(‘‘)

async def inner():
    LOG.info(‘inner 函数开始‘)
    time.sleep(0.1)
    LOG.info(‘inner 运行完成‘)

async def outer(loop):
    LOG.info(‘outer 函数开始‘)
    #ensure_future，直到await才运行
    await asyncio.ensure_future(loop.create_task(inner()))
    LOG.info(‘outer 运行完成‘)

event_loop = asyncio.get_event_loop()
if args.verbose:
    LOG.info(‘开启DEBUG模式‘)
    event_loop.set_debug(True)

    # 使“慢”任务的阈值非常小以便于说明。默认值为0.1，即100毫秒。
    event_loop.slow_callback_duration = 0.001

    # 在警告过滤器列表中插入一个简单的条目（在前面）。
    warnings.simplefilter(‘always‘, ResourceWarning)

LOG.info(‘entering event loop‘)
event_loop.run_until_complete(outer(event_loop))

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import asyncio
import logging
import sys

@asyncio.coroutine
def echo(reader, writer):
    address = writer.get_extra_info(‘peername‘)
    log = logging.getLogger(‘echo_{}_{}‘.format(*address))
    log.debug(‘connection accepted‘)
    while True:
        data = yield from reader.read(128)
        if data:
            log.debug(‘received {!r}‘.format(data))
            writer.write(data)
            yield from writer.drain()
            log.debug(‘sent {!r}‘.format(data))
        else:
            log.debug(‘closing‘)
            writer.close()
            return

#开启Debug模式
logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(name)s: %(message)s‘,
    stream=sys.stderr,
)
#设置日志的title
log = logging.getLogger(‘main‘)

#设置开启服务的IP+端口
server_address = (‘localhost‘, 8888)

#获取事件循环
event_loop = asyncio.get_event_loop()

# 创建服务器，让循环在之前完成协同工作。并且启动实际事件循环
coroutine = asyncio.start_server(echo, *server_address,loop=event_loop)
server = event_loop.run_until_complete(coroutine)
log.debug(‘starting up on {} port {}‘.format(*server_address))

try:
    #开启一直循环处理任务
    event_loop.run_forever()
finally:
    #结束后清理的工作
    log.debug(‘closing server‘)
    server.close()
    event_loop.run_until_complete(server.wait_closed())
    log.debug(‘closing event loop‘)
    event_loop.close()

import asyncio
import logging
import sys

logging.basicConfig(
    level=logging.DEBUG,
    format=‘%(name)s: %(message)s‘,
    stream=sys.stderr,
)
LOG = logging.getLogger(‘main‘)


async def stopper(loop):
    LOG.debug(‘stopper invoked‘)
    loop.stop()


event_loop = asyncio.get_event_loop()

event_loop.create_task(stopper(event_loop))

try:
    LOG.debug(‘entering event loop‘)
    event_loop.run_forever()
finally:
    LOG.debug(‘closing event loop‘)
    event_loop.close()