标签: python-asyncio

我是Python 3的新手,正在玩asyncio.因此,我遇到了以下服务器端代码的奇怪行为:

import asyncio


@asyncio.coroutine
def handle_client(reader, writer):
    print('Client connected.')
    client_connected = True
    while client_connected:
        print('Waiting for client event.')
        line = yield from reader.readline()
        if line:
            print('Got: {}'.format(line))
            if line.decode() == 'echo\n':
                print('Sending back echo.')
                writer.write(line)
            else:
                print('Not sending back anything.')
        else:
            print('Client disconnected.')
            client_connected = False


if __name__ == '__main__':
    asyncio.async(asyncio.start_server(handle_client, 'localhost', 8888))
    asyncio.get_event_loop().run_forever()

当我运行这个客户端代码时(编辑:客户端代码手动输入到IPython会话中,服务器肯定有时间在关闭套接字之前写入)...

import socket
client = socket.create_connection(('localhost', 8888))
client.sendall('echo\n'.encode())
client.close()

...我从服务器收到错误回溯:

C:\Users\Gnar\Anaconda3\python.exe C:/Users/Gnar/Code/echo.py
Client connected.
Waiting for client event.
Got: b'echo\n'
Sending back echo.
Waiting for …

我正在尝试使用asyncio/aioamqp编写异步使用者.我的问题是,回调协程(下面)是阻塞的.我将通道设置为执行basic_consume(),并将回调分配为callback().回调有一个"来自asyncio.sleep的产生"语句(模拟"工作"),它从发布者获取一个整数,并在打印消息之前休眠一段时间.

如果我发布了两条消息,一条时间为"10",紧接着一条时间为"1",我预计第二条消息将首先打印,因为它的睡眠时间较短.相反,回调阻止10秒,打印第一条消息,然后打印第二条消息.

似乎basic_consume或回调在某处阻塞.还有另外一种方法可以处理吗？

@asyncio.coroutine
def callback(body, envelope, properties):
    yield from asyncio.sleep(int(body))
    print("consumer {} recved {} ({})".format(envelope.consumer_tag, body, envelope.delivery_tag))

@asyncio.coroutine
def receive_log():
    try:
        transport, protocol = yield from aioamqp.connect('localhost', 5672, login="login", password="password")
    except:
        print("closed connections")
        return

    channel = yield from protocol.channel()
    exchange_name = 'cloudstack-events'
    exchange_name = 'test-async-exchange'
    queue_name = 'async-queue-%s' % random.randint(0, 10000)
    yield from channel.exchange(exchange_name, 'topic', auto_delete=True, passive=False, durable=False)
    yield from asyncio.wait_for(channel.queue(queue_name, durable=False, auto_delete=True), timeout=10)

    binding_keys = ['mykey']

    for binding_key in binding_keys:
        print("binding", binding_key)
        yield from asyncio.wait_for(channel.queue_bind(exchange_name=exchange_name,
                                                       queue_name=queue_name,
                                                       routing_key=binding_key), timeout=10) …

我正在设计一个Python应用程序,它应该访问一台机器来执行一些(冗长的)任务.对于与网络相关的所有内容,asyncio模块似乎是一个不错的选择,但现在我需要访问一个特定组件的串行端口.我已经为实际的串口实现了一种抽象层,但是无法弄清楚如何将它与asyncio明智地集成在一起.

下面的设置:我有一个运行循环的线程,它经常与机器对话并解码响应.使用一种方法enqueue_query(),我可以将一个查询字符串放入一个队列,然后由另一个线程将其发送到机器并引发响应.通过传入threading.Event(或带有set()方法的任何东西),调用者可以执行阻塞等待响应.这可能看起来像这样:

f = threading.Event()
ch.enqueue_query('2 getnlimit', f)
f.wait()
print(ch.get_query_responses())

我的目标是将这些行放入协程并让asyncio处理这个等待,以便应用程序可以在此期间执行其他操作.我怎么能这样做？它可能会通过将其包装f.wait()到Executor中来实现,但这似乎有点愚蠢,因为这会创建一个新线程,只是等待另一个线程做某事.

谢谢!最好的问候,菲利普

我写了一个测试程序来尝试使用create_task(),需要等到创建的任务完成.

我尝试使用loop.run_until_complete()等待任务完成,但它导致跟踪错误.

/Users/jason/.virtualenvs/xxx/bin/python3.5 /Users/jason/asyncio/examples/hello_coroutine.py
    Traceback (most recent call last):
    Test
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 42, in <module>
    Hello World, is a task
        loop.run_until_complete(test.greet_every_two_seconds())
      File "/Users/jason/asyncio/asyncio/base_events.py", line 373, in run_until_complete
        return future.result()
      File "/Users/jason/asyncio/asyncio/futures.py", line 274, in result
        raise self._exception
      File "/Users/jason/asyncio/asyncio/tasks.py", line 240, in _step
        result = coro.send(None)
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 33, in greet_every_two_seconds
        self.a()
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 26, in a
        t = loop.run_until_complete(self.greet_every_one_seconds(self.db_presell))
      File "/Users/jason/asyncio/asyncio/base_events.py", line 361, in run_until_complete
        self.run_forever()
      File "/Users/jason/asyncio/asyncio/base_events.py", line 326, in run_forever …

我有这个代码:

try:
    asyncio.ensure_future(data_streamer.sendByLatest())
except ValueError as e:
    logging.debug(repr(e))

data_streamer.sendByLatest()可以提高ValueError,但它没有被抓住.

我是Python的新手.我用asyncio编写了一个简单的报废程序.这是我的代码片段

 loop = asyncio.get_event_loop()
 task = loop.create_task(conSpi.parse(arguments.url))
 value = loop.run_until_complete(asyncio.wait([task]))
 loop.close()

我想打印返回值的结果.收集打印变量的值,打印出类似这样的内容

 {<Task finished coro=<ConcurrentSpider.parse() done, 
 defined at /home/afraz/PycharmProjects/the-lab/concurrentspider.py:28> result=3>}

`

我怎样才能得到结果而不打印？

我想在我的Python库中使用select.epoll().

不幸的是epoll并不是随处可见的.

我需要一种方法来回退到select.select().

我试图在pypi找到一些东西,但未能找到匹配的包:https://pypi.python.org/pypi？%3Aaction = search&term = epoll&subsmit = search

我怎么能解决这个"从epoll回退选择是否epoll不可用"？

我在模拟aiohttp.client.ClientSession.get上下文管理器时遇到了一些麻烦.我找到了一些文章,这里有一个似乎正在起作用的例子:第1条

所以我要测试的代码:

async_app.py

import random
from aiohttp.client import ClientSession

async def get_random_photo_url():
    while True:
        async with ClientSession() as session:
            async with session.get('random.photos') as resp:
                json = await resp.json()
        photos = json['photos']
        if not photos:
            continue
        return random.choice(photos)['img_src']

并测试:

test_async_app.py

from asynctest import CoroutineMock, MagicMock, patch

from asynctest import TestCase as TestCaseAsync

from async_app import get_random_photo_url


class AsyncContextManagerMock(MagicMock):
    async def __aenter__(self):
        return self.aenter

    async def __aexit__(self, *args):
        pass

class TestAsyncExample(TestCaseAsync):
    @patch('aiohttp.client.ClientSession.get', new_callable=AsyncContextManagerMock)
    async def test_call_api_again_if_photos_not_found(self, mock_get):
        mock_get.return_value.aenter.json = CoroutineMock(side_effect=[{'photos': []}, …

在Python调试器中是否可以await任意调用async函数？

说我在某些main.py文件中有以下代码：

import asyncio

async def bar(x):
    return x + 1

async def foo():
    import ipdb; ipdb.set_trace()

asyncio.run(foo())

现在，我想bar()在调试器中测试带有某些参数的调用以测试结果。发生以下情况：

$ python3 main.py
> /Users/user/test/main.py(8)foo()
      7     import ipdb; ipdb.set_trace()
----> 8     return None
      9

ipdb> bar(1)
<coroutine object bar at 0x10404ae60>
main.py:1: RuntimeWarning: coroutine 'bar' was never awaited
  import asyncio
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
ipdb> await bar(1)
*** SyntaxError: 'await' outside function

当然，我可以通过在自己的x = await bar(1)之上 …

我尝试从 Binance Websocket 获取数据。使用 python 3.9 作为解释器它运行良好，但使用 3.10 它给了我错误:(

这是代码：

import asyncio
from binance import AsyncClient, BinanceSocketManager


async def main():
    client = await AsyncClient.create()
    bm = BinanceSocketManager(client)
    # start any sockets here, i.e a trade socket
    ts = bm.trade_socket('BNBBTC')
    # then start receiving messages
    async with ts as tscm:
        while True:
            res = await tscm.recv()
            print(res)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

我收到此错误：

DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()

我使用 PyCharm 作为 IDE。

请问有人可以帮助我吗？