import asyncio
# 用async定義一個(gè)協(xié)程
async def wait_and_print(wait_time, name):
    # 這是一個(gè)模擬io阻塞的定時(shí)器，在sleep(或者io)的時(shí)候，
    # 協(xié)程會(huì)把控制權(quán)交還給event_loop,讓他去執(zhí)行別的協(xié)程
    await asyncio.sleep(wait_time)
    print(f'wait_time:{wait_time}, name:{name}')
# 直接 wait_and_print(2, 'a') 這樣不能運(yùn)行一個(gè)協(xié)程，要將它注冊(cè)入事件循環(huán)中
# 在python3.7
# asyncio.run(wait_and_print(2, 'a'))
# 在python3.6
loop = asyncio.get_event_loop()
# run_until_complete 這個(gè)函數(shù)會(huì)阻塞運(yùn)行直到里面的協(xié)程運(yùn)行完畢返回結(jié)果
loop.run_until_complete(wait_and_print(2, 'a'))

import asyncio
async def wait_and_print(wait_time, name):
    # await會(huì)阻塞直到后面跟的協(xié)程運(yùn)行完畢返回結(jié)果
    await asyncio.sleep(wait_time)
    print(f'wait_time:{wait_time}, name:{name}')
# # 1.直接注冊(cè)入事件循環(huán)
# loop = asyncio.get_event_loop()
# loop.run_until_complete(wait_and_print(2, 'a'))
#
#
# # 2.間接注冊(cè)入事件循環(huán)
# # 在其他協(xié)程中await它，再把那個(gè)協(xié)程注冊(cè)入事件循環(huán)
# async def run():
#     await wait_and_print(2, 'a')
#
#
# loop = asyncio.get_event_loop()
# loop.run_until_complete(run())
# # 3.打包成Task對(duì)象
#
async def run():
    # 關(guān)于下面兩個(gè)函數(shù)的區(qū)別后面會(huì)在說完Task和Future對(duì)象后進(jìn)行討論
    # python3.6
    # asyncio.ensure_future(wait_and_print(2, 'a'))
    # python3.7
    print(asyncio.create_task(wait_and_print(2, 'a')))
    await asyncio.sleep(3)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())

import asyncio
async def wait_and_print(wait_time, name):
    await asyncio.sleep(wait_time)
    print(f'wait_time:{wait_time}, name:{name}')
async def run():
    # 為了我們方便查看這個(gè)任務(wù)的后續(xù)狀態(tài)，先打包成一個(gè)task_
    task_ = asyncio.create_task(wait_and_print(1000, 'a'))
    try:
        await asyncio.wait_for(task_, timeout=3)
    except asyncio.TimeoutError:
        print('發(fā)現(xiàn)任務(wù)超時(shí)，已被取消')
        # 查看被取消后的狀態(tài)
        print(task_)
        print(task_.cancelled())
asyncio.run(run())

import asyncio
async def wait_and_print(wait_time, name):
    await asyncio.sleep(wait_time)
    print(f'wait_time:{wait_time}, name:{name}')
    # 添加一個(gè)return
    return f'wait_time:{wait_time}, name:{name}'
# loop = asyncio.get_event_loop()
# # 會(huì)把三個(gè)協(xié)程打包成任務(wù)task加入循環(huán)
# gather_result = loop.run_until_complete(asyncio.gather(wait_and_print(1, 'a'),
#                                                        wait_and_print(2, 'b'),
#                                                        wait_and_print(1, 'c')))
# print(gather_result)
async def wait_and_cancel_task(wait_time, task):
    await asyncio.sleep(wait_time)
    task.cancel()
    print(f'{task} 被取消了')
    return f'{task} 被取消了'
async def run():
    # 封裝一個(gè)b_task
    b_task = asyncio.create_task(wait_and_print(5, 'b'))
    print(b_task)
    gather_ = asyncio.gather(wait_and_print(1, 'a'),
                             b_task,
                             wait_and_print(3, 'c'),
                             # 在這里把他給取消了
                             wait_and_cancel_task(4, b_task), return_exceptions=True)
    print(await gather_)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())

   import asyncio
   async def compute(x, y):
       print("Compute %s + %s ..." % (x, y))
       await asyncio.sleep(1.0)
       return x + y
   async def print_sum(x, y):
       result = await compute(x, y)
       print("%s + %s = %s" % (x, y, result))
   loop = asyncio.get_event_loop()
   loop.run_until_complete(print_sum(1, 2))
   loop.close()

import asyncio
async def wait_and_print(wait_time, name):
    await asyncio.sleep(wait_time)
    print(f'wait_time:{wait_time}, name:{name}')
async def run():
    asyncio.create_task(wait_and_print(1, 'a'))
    await wait_and_print(1, 'b')
if __name__ == '__main__':
    asyncio.run(run())
# 原因：create_task會(huì)把任務(wù)自動(dòng)安排到下一輪事件循環(huán)

import asyncio
async def wait_and_print(wait_time, name):
    await asyncio.sleep(wait_time)
    print(f'wait_time:{wait_time}, name:{name}')
async def run():
    asyncio.create_task(wait_and_print(1, 'a'))
    await asyncio.sleep(0)
    await wait_and_print(1, 'b')
if __name__ == '__main__':
    asyncio.run(run())
# 原因：await asyncio.sleep(0)會(huì)使得task將控制權(quán)交還給loop進(jìn)行下一個(gè)任務(wù)

import asyncio
async def wait_and_print(wait_times, name):
    await asyncio.sleep(wait_times)
    print(f'wait_times:{wait_times}  name:{name}')
async def run():
    asyncio.ensure_future(asyncio.ensure_future(wait_and_print(1, 'f')))
    asyncio.ensure_future(wait_and_print(1, 'b'))
    asyncio.ensure_future(asyncio.gather(wait_and_print(1, 'd')))
    asyncio.create_task(wait_and_print(1, 'a'))
    await wait_and_print(1, 'c')
    asyncio.ensure_future(asyncio.gather(wait_and_print(2, 'g'), wait_and_print(0, 'h')))
    await wait_and_print(1, 'e')
# c>f>b>d>a>h>e>
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run())

import asyncio
import random
async def mama(zhuo_zi):
    while True:
        zheng_bao_zi = random.randint(1, 10)
        await asyncio.sleep(zheng_bao_zi)
        pan_zi = random.randint(1, 10)
        [await zhuo_zi.put(bao_zi) for bao_zi in range(pan_zi)]
        print(f'媽媽花了{(lán)zheng_bao_zi}秒，上桌了{(lán)pan_zi}個(gè)包子')
        print(f'現(xiàn)在桌子上的包子總數(shù):{zhuo_zi.qsize()}')
async def xiao_ming(zhuo_zi):
    while True:
        if zhuo_zi.qsize() == 0:
            print(f'小明發(fā)現(xiàn)沒包子了，等一會(huì)')
            await asyncio.sleep(1)
        else:
            await zhuo_zi.get()
            chi_bao_zi = random.randint(1, 3)
            await asyncio.sleep(chi_bao_zi)
            print(f'小明花了{(lán)chi_bao_zi}秒吃了個(gè)包子')
async def run(mama_number, xiao_ming_number):
    zhuo_zi = asyncio.Queue()
    mama_tasks = [asyncio.create_task(mama(zhuo_zi)) for number in range(mama_number)]
    xiao_ming_tasks = [asyncio.create_task(xiao_ming(zhuo_zi)) for number in range(xiao_ming_number)]
    return mama_tasks, xiao_ming_tasks
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    mama_tasks, xiao_ming_tasks = loop.run_until_complete(run(mama_number=3, xiao_ming_number=8))
    loop.run_forever()

import asyncio
import logging
import time
import json
log = logging
def parsed_response(response):
    """
       解析 response
       :param : response byte
       :return: status_code int
                headers dict
                str body
    """
    r = response.decode()
    header, body = r.split('\r\n\r\n', 1)
    h = header.split('\r\n')
    status_code = int(h[0].split()[1])
    headers = {}
    for line in h[1:]:
        k, v = line.split(': ')
        headers[k] = v
    return status_code, headers, body
def parse_url(url):
    """
    對(duì) url 進(jìn)行字符串拆分
    :param url: http://httpbin.org/headers
    :return: (protocol, host, port, path)
    """
    # 檢查協(xié)議
    protocol = 'http'
    if url[:7] == 'http://':
        u = url.split('://')[1]
    elif url[:8] == 'https://':
        protocol = 'https'
        u = url.split('://')[1]
    else:
        u = url
    # https://g.cn:1234/hello
    # g.cn:1234/hello
    # 檢查默認(rèn) path
    i = u.find('/')
    if i == -1:
        host = u
        path = '/'
    else:
        host = u[:i]
        path = u[i:]
    # 檢查端口
    port_dict = {
        'http': 80,
        'https': 443,
    }
    # 默認(rèn)端口
    port = port_dict[protocol]
    if ':' in host:
        h = host.split(':')
        host = h[0]
        port = int(h[1])
    return protocol, host, port, path
def save_response(queue, save_name='data'):
    """
    將一個(gè)隊(duì)列里的數(shù)據(jù)保存到j(luò)son文本
    :param queue: asyncio.Queue
    :param save_name: str json文本的name
    :return:
    """
    data = []
    for i in range(queue.qsize()):
        status_code, headers, body = queue.get_nowait()
        save_dict = dict(id=i,
                         status_code=status_code,
                         headers=headers,
                         body=body)
        data.append(save_dict)
    s = json.dumps(data, indent=2, ensure_ascii=False)
    with open(save_name, 'w', encoding='utf-8') as f:
        f.write(s)
async def connect(host, port, path='headers', tls=False):
    """
    :param host: str: httpbin.org
    :param port: int: 80/443
    :param tls:  bool
    :param path: str: headers/
    :return: response content (bytes)
    """
    log.debug(f'connect to {host}:{port}, tls:{tls}')
    if tls is True:
        protocol = 'https'
        reader, writer = await asyncio.open_connection(
            host,
            port,
            ssl=True)
    else:
        protocol = 'http'
        reader, writer = await asyncio.open_connection(
            host,
            port)
    log.debug(f'{host}:{port}, tls:{tls} connected')
    request = f'GET {protocol}://{host}:{port}/{path} HTTP/1.1\r\nHost:{host}\r\n\r\n'
    # 由于設(shè)置了timeout, 在以下操作任意階段皆有可能超時(shí)導(dǎo)致協(xié)程任務(wù)被丟棄
    # 所以必須添加finally字段用來正確關(guān)閉連接
    try:
        log.debug(f'request send--{request}')
        writer.write(request.encode('utf-8'))
        await writer.drain()
        log.debug(f'request send done--{request}')
        log.debug(f'response recv start')
        response = b''
        # 接收response
        while True:
            line = await reader.readline()
            response += line
            if line == b'}\n':
                break
        log.debug(f'response recv done')
        log.debug(f'close connect')
        writer.close()
        return response
    finally:
        writer.close()
async def http_request(url):
    """
    返回請(qǐng)求后的狀態(tài)碼，頁面內(nèi)容
    :param url: http://httpbin.org/headers
    :return: state_code, header, body
    """
    protocol, host, port, path = parse_url(url)
    if port == 443:
        response = await connect(host, port, tls=True)
    else:
        response = await connect(host, port, tls=False)
    status_code, headers, body = parsed_response(response)
    return status_code, headers, body
async def create_one_coroutine(url_queue, coros_queue, response_queue, task_timeout=1):
    """
    :param url_queue:
    :param coros_queue:
    :param response_queue:
    :param task_timeout:
    :return:
    """
    # 用當(dāng)前協(xié)程池隊(duì)列大小給予協(xié)程id
    thread_id = coros_queue.qsize()
    log.info(f'添加一個(gè)協(xié)程{thread_id}')
    # 將協(xié)程放入?yún)f(xié)程池隊(duì)列
    await coros_queue.put(thread_id)
    while True:
        # 如果urls任務(wù)隊(duì)列空了，則從協(xié)程池隊(duì)列中去掉一個(gè)協(xié)程
        if url_queue.empty() is True:
            await coros_queue.get()
            log.info(f'協(xié)程{thread_id}關(guān)閉')
            break
        url = await url_queue.get()
        log.info(f'協(xié)程{thread_id}從隊(duì)列中拿取一個(gè)url, 還剩{url_queue.qsize()}個(gè)url')
        try:
            # 將http_request(url)返回的response裝入response隊(duì)列中
            await response_queue.put(
                await asyncio.wait_for(http_request(url), timeout=task_timeout))
        # asyncio.wait_for()超時(shí)會(huì)引發(fā)一個(gè)TimeoutError
        except asyncio.TimeoutError:
            log.info(f'協(xié)程{thread_id}請(qǐng)求超時(shí)，丟棄這次請(qǐng)求')
        else:
            log.info(f'協(xié)程{thread_id}成功獲取一次響應(yīng)')
async def if_threads_done(threads_queue, response_queue):
    """
    監(jiān)視urls協(xié)程的管理協(xié)程,
    每1秒對(duì)urls協(xié)程隊(duì)列進(jìn)行一次判定，
    如果urls協(xié)程隊(duì)列已空則關(guān)閉loop
    :param response_queue: asyncio.Queue
    :param threads_queue: asyncio.Queue
    :return:
    """
    start_time = time.time()
    while True:
        # 每隔一秒檢查一次協(xié)程池隊(duì)列是否為空
        await asyncio.sleep(1)
        if threads_queue.empty() is True:
            log.info(f'所有協(xié)程已經(jīng)關(guān)閉，關(guān)閉loop')
            break
    # 如果協(xié)程池隊(duì)列已經(jīng)沒有協(xié)程在運(yùn)行，則統(tǒng)計(jì)時(shí)間->保存responses->關(guān)閉loop
    log.info(f'獲取成功的響應(yīng)數(shù)量:{response_queue.qsize()}')
    save_response(response_queue, save_name='response.txt')
    use_time = time.time() - start_time
    log.info(f'總耗時(shí):{str(use_time)}')
    log.info(f'響應(yīng)已存入response.txt')
    loop = asyncio.get_running_loop()
    loop.stop()
async def run_with_pool(urls, threads=5, task_timeout=5):
    """
    :param urls: list
    :param threads: int
    :param task_timeout: int
    :return: loop
    """
    # 創(chuàng)建一個(gè)url隊(duì)列，用于協(xié)程提取url
    url_queue = asyncio.Queue()
    # 創(chuàng)建一個(gè)coroutines隊(duì)列，用于管理協(xié)程
    coros_queue = asyncio.Queue()
    # 創(chuàng)建一個(gè)response隊(duì)列，用于保存response
    response_queue = asyncio.Queue()
    # 將urls列表加入隊(duì)列
    [url_queue.put_nowait(url) for url in urls]
    # 創(chuàng)建協(xié)程池
    coros = [create_one_coroutine(url_queue, coros_queue, response_queue, task_timeout)
             for i in range(0, threads)]
    [asyncio.create_task(coro) for coro in coros]
    # 創(chuàng)建一個(gè)監(jiān)視urls協(xié)程的管理協(xié)程， 每1秒對(duì)urls協(xié)程隊(duì)列進(jìn)行一次判定，如果urls協(xié)程隊(duì)列已空則關(guān)閉loop
    asyncio.create_task(if_threads_done(coros_queue, response_queue))
def main():
    tasks = [
                'http://httpbin.org'
            ] * 50
    tasks += [
                 'https://httpbin.org'
             ] * 50
    loop = asyncio.get_event_loop()
    loop.create_task(run_with_pool(tasks, threads=15, task_timeout=3))
    try:
        loop.run_forever()
    finally:
        loop.close()
if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    main()