标签: python-asyncio

根据文档,有几种Queues实现.我感兴趣的是JoinableQueue和PriorityQueue,因为我想拥有一个具有优先级的可连接队列.

似乎我只能获得较低版本中的一个功能.即3.5我可以使用Queue(可连接)和PriorityQueue进行区分,但是在3.5以下的python中有JoinableQueue和PriorityQueue(参见下面的例子).

是否可以将它们组合起来以获得在3.4中获得可连接的PriorityQueue的一般方法？

try:
    # Python 3.4.
    from asyncio import JoinableQueue as Queue # joinable
    from asyncio import PriorityQueue          # I assume this one is not joinable
except ImportError:
    # Python 3.5.
    from asyncio import Queue                  # standard joinable
    from asyncio import PriorityQueue          # I assume this is the one I want

另一种方法是以某种方式影响队列？

import asyncio
import aiohttp
import bs4
import tqdm


@asyncio.coroutine
def get(*args, **kwargs):
    response = yield from aiohttp.request('GET', *args, **kwargs)
    return (yield from response.read_and_close(decode=True))


@asyncio.coroutine
def wait_with_progress(coros):
    for f in tqdm.tqdm(asyncio.as_completed(coros), total=len(coros)):
        yield from f


def first_magnet(page):
   soup = bs4.BeautifulSoup(page)
   a = soup.find('a', title='Download this torrent using magnet')
   return a['href']


@asyncio.coroutine
def print_magnet(query):
    url = 'http://thepiratebay.se/search/{}/0/7/0'.format(query)
    with (yield from sem):
        page = yield from get(url, compress=True)
    magnet = first_magnet(page)
    print('{}: {}'.format(query, magnet))

distros = ['archlinux', 'ubuntu', 'debian']
sem = asyncio.Semaphore(5)
loop …

我需要有一个长期运行的websocket客户端,它接收来自websocket服务器的推送消息,我需要监控客户端的连接状态:如果连接断开,我需要找出答案.

我的方法是定期记录一个常量字符串,并在未检测到日志消息时触发警报.

我的想法:1)有一个websocket客户端响应不规则传入的消息.2)同时有一个循环,当websocket客户端抛出一个ConnectionClosed执行时停止记录消息.

我对新的3.5异步语法很感兴趣.这个websocket实现特别基于asyncio.文档中的客户端看起来与我需要的完全一样.

但是,我不知道如何添加第二个coroutine来执行我的日志记录语句,并在websocket连接抛出ConnectionClosed时以某种方式停止.

这是开始对话的东西,但这不起作用,因为alive方法阻止事件循环.我正在寻找的是同时运行两种方法的优雅解决方案.

#!/usr/bin/env python

import asyncio
import logging

import websockets

logger = logging.getLogger(__name__)

is_alive = True


async def alive():
    while is_alive:
        logger.info('alive')
        await asyncio.sleep(300)


async def async_processing():
    async with websockets.connect('ws://localhost:8765') as websocket:
        while True:
            try:
                message = await websocket.recv()
                print(message)

            except websockets.exceptions.ConnectionClosed:
                print('ConnectionClosed')
                is_alive = False
                break


asyncio.get_event_loop().run_until_complete(alive())
asyncio.get_event_loop().run_until_complete(async_processing())

aiohttpwebsocket支持（即WebSocketResponse）和websockets （均提供异步支持）有什么区别？可以混合吗？

我有一个基于异步的类，我想对其进行单元测试。使用tornado.testing.AsyncTestCase此工具非常容易。但是，我班上的一个特定方法asyncio.ensure_future用来安排另一个方法的执行。这永远不会在中完成AsyncTestCase，因为默认的测试运行器使用龙卷风KQueueIOLoop事件循环，而不是asyncio事件循环。

class TestSubject:
    def foo(self):
        asyncio.ensure_future(self.bar())

    async def bar(self):
        pass

class TestSubjectTest(AsyncTestCase):
    def test_foo(self):
        t = TestSubject()
        # here be somewhat involved setup with MagicMock and self.stop
        t.foo()
        self.wait()

$ python -m tornado.testing baz.testsubject_test
...
[E 160627 17:48:22 testing:731] FAIL
[E 160627 17:48:22 base_events:1090] Task was destroyed but it is pending!
    task: <Task pending coro=<TestSubject.bar() running at ...>>
.../asyncio/base_events.py:362: RuntimeWarning: coroutine 'TestSubject.bar' was never awaited

如何使用其他事件循环来运行测试以确保我的任务将被实际执行？或者，如何使我的实现事件独立于循环和交叉兼容？

我从websockets开始，并且asyncio很难理解背后的逻辑。过去，我在Python上使用过线程，而在Javascript上使用过AJAX调用（因此，我对多线程（绑定了I / O）和可异步执行结果的过程的概念有所了解）。

示例服务器如下。我的目的是提供一个websocket连接和两个函数，每10秒将调用一次，而无需等待结果（因此，每10秒恰好一个调用，而不管这些函数要花多长时间）。

import time
import arrow
import websockets
import asyncio

async def time1(websocket):
    # simulation of a time intensive operation
    time.sleep(2)
    msg = '1 ' + arrow.now()
    print(msg)
    websocket.send(msg)

async def time2(websocket):
    # simulation of a time intensive operation
    time.sleep(3)
    msg = '2 ' + arrow.now()
    print(msg)
    websocket.send(msg)

async def update_info(websocket, path):
    while True:
        now = arrow.now()
        if not now.second % 10:
            time1(websocket)
        if not now.second % 10:
            time2(websocket)

        time.sleep(1)

start_server = websockets.serve(update_info, 'localhost', 5678)
asyncio.get_event_loop().run_until_complete(start_server) …

假设我有两个异步生成器:

async def get_rules():
    while True:
        yield 'rule=1'
        asyncio.sleep(2)


async def get_snapshots():
    while True:
        yield 'snapshot=1'
        asyncio.sleep(5)

我想将它们合并到一个异步生成器中,该生成器返回2元组,其中包含两个元组的最新值.排序combineLatest.

做这个的最好方式是什么？

我只是不知道如何通过阅读aiohttp示例或google 来重用aiomysql连接池。这是我的代码

import aiomysql
import asyncio


async def select(loop, sql):
    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,
                                      user='root', password='123456',
                                      db='test', loop=loop)
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute(sql)
            r = await cur.fetchone()
            print(r)


async def insert(loop, sql):
    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,
                                      user='root', password='123456',
                                      db='test', loop=loop)
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute(sql)
            await conn.commit()


async def main(loop):
    c1 = select(loop=loop, sql='select * from minifw')
    c2 = insert(loop=loop, sql="insert into …

我正在尝试使用python-socketio设置socket.io服务器。这是一个最小的工作示例：

import asyncio
from aiohttp import web
import socketio
import random

sio = socketio.AsyncServer(async_mode='aiohttp')
app = web.Application()
sio.attach(app)

@sio.on('connect')
def connect(sid, environ):
    print("connected: ", sid)

@sio.on('sendText')
async def message(sid, data):
    print("message ", data)
    # await asyncio.sleep(1 * random.random())
    # print('waited', data)

@sio.on('disconnect')
def disconnect(sid):
    print('disconnect ', sid)

if __name__ == '__main__':
    web.run_app(app, host='0.0.0.0', port=8080)

运行正常，例如，我可以执行（在node.js中）

const io = require('socket.io-client');
const socket = io('ws://localhost:8080');
socket.emit('sendText', 'hey 1')
socket.emit('sendText', 'hey 2')
socket.emit('sendText', 'hey 3')

如果我运行服务器并运行上面的节点脚本，我将获得服务器端

已连接：c1e687f0e2724b339fcdbefdb5aaa8f8

消息嘿1

消息嘿2

消息嘿3

但是，如果我在代码中用await …

我必须发送大量HTTP请求,一旦所有HTTP请求都返回,程序就可以继续.听起来像是一场完美的比赛asyncio.有点天真,我把我的电话包裹requests在一个async函数中,然后把它们给了asyncio.这不起作用.

在线搜索后,我找到了两个解决方案:

• 使用像aiohttp这样的库,可以使用它asyncio
• 在调用中包装阻塞代码 run_in_executor

为了更好地理解这一点,我写了一个小基准.服务器端是一个烧瓶程序,在回答请求之前等待0.1秒.

from flask import Flask
import time

app = Flask(__name__)


@app.route('/')
def hello_world():
    time.sleep(0.1) // heavy calculations here :)
    return 'Hello World!'


if __name__ == '__main__':
    app.run()

客户是我的基准

import requests
from time import perf_counter, sleep

# this is the baseline, sequential calls to requests.get
start = perf_counter()
for i in range(10):
    r = requests.get("http://127.0.0.1:5000/")
stop = perf_counter()
print(f"synchronous took {stop-start} seconds") # 1.062 secs …