标签: python-asyncio

我想在app.py上创建两个带有asyncio的协议(TcpClient和UdpServer),其中TcpClient将与server.py和UdpServer作为UDP服务器建立持久连接:

我需要的是:
a)两种通信协议:相互调用方法.这只是在第一个连接上工作.如果TcpClient重新连接,则无法再次发送字符串"send to tcp".来自UdpServer.我检查print(self)并且TcpClient创建一个新实例,旧的仍然存在但没有连接,但我不知道如何重构.我认为我以错误的方式使用asyncio.
b)当TcpClient与server.py断开连接时,等待5s并再次尝试重新连接,依此类推.我试图使用call_later()asyncio,但我认为有一种本地方式来做到这一点,而不是一种技巧.
c)当我启动app.py并且如果TcpClient无法连接时,我想尝试在5s后重新连接,依此类推.我不知道该怎么做.

这是我的app.py server.py的示例测试.server.py仅用于测试 - 这将是另一种语言.

只是说我尝试了什么:
1)当我启动app.py并且server.py关闭时,app.py不会重试.
2)当app.py连接到server.py并且服务器关闭并快速启动时,TcpClient重新连接,但我无法在新实例上相互连接方法并发送字符串"send to tcp".到server.py,只是旧的,没有更多的连接.
3)如果我用asyncio.async(),而不是run_until_complete()我不能打电话相互协议的方法.

我把app.py和server.py放在这里,所以你可以复制并运行测试.

我ncat localhost 9000 -u -v用来发送字符串"send to tcp.".此字符串需要打印在UdpServer类上,并传递给TcpClient类上的方法send_data_to_tcp,此方法将字符串发送到server.py.< - 首次重新连接tcpClient后,这不起作用.

我正在使用Python 3.4.0.

非常感谢.

app.py:

import asyncio

#TCP client
class TcpClient(asyncio.Protocol):
    message = 'Testing'

    def connection_made(self, transport):
        self.transport = transport
        self.transport.write(self.message.encode())
        print('data sent: {}'.format(self.message))
        server_udp[1].tcp_client_connected()


    def data_received(self, data):
        self.data = format(data.decode())
        print('data received: {}'.format(data.decode()))
        if self.data == 'Testing':
            server_udp[1].send_data_to_udp(self.data)

    def send_data_to_tcp(self, data):
        self.transport.write(data.encode())

    def …

我有一个函数download_all,它遍历一个硬编码的页面列表,按顺序下载它们.但是如果我想根据页面的结果动态添加到列表中,我该怎么办呢？例如,下载第一页,解析它,并根据结果将其他页面添加到事件循环中.

@asyncio.coroutine
def download_all():
    first_page = 1
    last_page = 100
    download_list = [download(page_number) for page_number in range(first_page, last_page)]
    gen = asyncio.wait(download_list)
    return gen

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    futures = loop.run_until_complete(download_all())

我正在使用aiohttp在python 3.4中创建一个简单的HTTP请求,如下所示:

response = yield from aiohttp.get(url)

应用程序一遍又一遍地请求相同的URL,所以我自然想要缓存它.我的第一次尝试是这样的:

@functools.lru_cache(maxsize=128)
def cached_request(url):
    return aiohttp.get(url)

第一次调用cached_request工作正常,但在以后的调用中我最终得到None而不是响应对象.

我对asyncio比较新,所以我尝试了很多asyncio.coroutine装饰器的组合,yield from以及其他一些东西,但似乎都没有.

那么缓存协同程序如何工作？

因为Python 3.5引入了async with在推荐的语法文档的aiohttp改变.现在要获得一个网址,他们建议:

import aiohttp
import asyncio

async def fetch(session, url):
    with aiohttp.Timeout(10):
        async with session.get(url) as response:
            return await response.text()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    with aiohttp.ClientSession(loop=loop) as session:
        html = loop.run_until_complete(
            fetch(session, 'http://python.org'))
        print(html)

如何修改此设置以获取网址集合而不仅仅是一个网址？

在旧asyncio示例中,您将设置一个任务列表,例如

    tasks = [
            fetch(session, 'http://cnn.com'),
            fetch(session, 'http://google.com'),
            fetch(session, 'http://twitter.com')
            ]

我试图将这样的列表与上面的方法结合起来但是失败了.

我正在尝试为使用"async with"语句的方法编写测试(在这种情况下,aioredis的连接池),我想模拟与redis的连接,但我无法弄清楚如何.

这是我到目前为止所拥有的:

from asyncio import Future
from unittest.mock import MagicMock

import pytest

# The thing i'm trying to test
async def set_value(redis, value):
    # Do things
    async with redis.get() as conn:
        await conn.set("key", value)

#My Mock classes
class MockRedis():
    def get(self):
        return MockAsyncPool()


class MockAsyncPool(MagicMock):
    async def __aenter__(self):
        conn = MagicMock()
        f = Future()
        f.set_result(True)
        conn.set = MagicMock(return_value=f)
        return conn

    def __aexit__(self, exc_type, exc_val, exc_tb):
        pass


# The actual test
@pytest.mark.asyncio
async def test_get_token():
    redis = MockRedis()

    token = …

这个线程中的 aiohttp 服务器示例失败并显示RuntimeError: There is no current event loop in thread 'Thread-1'.错误：

import threading
from aiohttp import web


def aiohttp_server():
    def say_hello(request):
        return web.Response(text='Hello, world')

    app = web.Application(debug=True)
    app.add_routes([web.get('/', say_hello)])
    web.run_app(app)


t = threading.Thread(target=aiohttp_server)
t.start()

如何在线程中运行 aiohttp 服务器？

我需要编写两个程序,这些程序将作为父进程及其子进程运行.父进程生成子进程,然后通过连接到子进程stdin和stdout的一对管道进行通信.通信是点对点的,这就是我需要asyncio的原因.一个简单的读/回放循环是行不通的.

我写过父母.没问题,因为asyncio我提供了所需的一切create_subprocess_exec().

但是我不知道如何在孩子中创建类似的流读取器/写入器.我没想到会有任何问题.因为已经创建了管道,并且在子进程启动时可以使用文件描述符0和1.没有连接是打开的,不需要生成任何进程.

这是我不努力的尝试:

import asyncio
import sys

_DEFAULT_LIMIT = 64 * 1024

async def connect_stdin_stdout(limit=_DEFAULT_LIMIT, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader(limit=limit, loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    r_transport, _ = await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    w_transport, _ = await loop.connect_write_pipe(lambda: protocol, sys.stdout)
    writer = asyncio.StreamWriter(w_transport, protocol, reader, loop)
    return reader, writer

问题是我有两个运输工具,我应该有一个.该函数失败,因为它尝试将协议的传输设置两次:

await loop.connect_read_pipe(lambda: protocol, sys.stdin)
await loop.connect_write_pipe(lambda: protocol, sys.stdout)
# !!!! assert self._transport is None, 'Transport already set'

我试图将伪协议传递给第一行,但这行也不正确,因为需要两个传输,而不仅仅是一个:

writer = …

是一个函数，如：

async def f(x):
    time.sleep(x)

await f(5)

正确异步/非阻塞？

asyncio 提供的睡眠功能有什么不同吗？

最后，aiorequests 是请求的可行异步替代品吗？

（在我看来，它基本上将主要组件包装为异步）

https://github.com/pohmelie/aiorequests/blob/master/aiorequests.py

鉴于此示例 Starlette 应用程序具有开放的 websocket 连接，您如何关闭 Starlette 应用程序？我在 uvicorn 上运行。每当我按下Ctrl+C输出时，Waiting for background tasks to complete.它就会永远挂起。

from starlette.applications import Starlette

app = Starlette()

@app.websocket_route('/ws')
async def ws(websocket):
    await websocket.accept()

    while True:
        # How to interrupt this while loop on the shutdown event?
        await asyncio.sleep(0.1)

    await websocket.close()

我尝试在关闭事件上切换 bool 变量，但该变量永远不会更新。它总是False。

例如。

app.state.is_shutting_down = False


@app.on_event('shutdown')
async def shutdown():
    app.state.is_shutting_down = True


@app.websocket_route('/ws')
async def ws(websocket):
    await websocket.accept()

    while app.state.is_shutting_down is False:

我正在尝试创建一个 WebSocket 命令行客户端，它等待来自 WebSocket 服务器的消息但同时等待用户输入。

每秒定期轮询多个在线资源在服务器上运行良好，（localhost:6789在本示例中运行的那个），但sleep()它使用而不是使用 Python 的正常方法asyncio.sleep()，这是有道理的，因为睡眠和异步睡眠不是一回事，在至少不是在引擎盖下。

同样，等待用户输入和异步等待用户输入不是一回事，但我无法弄清楚如何以异步等待任意秒数的方式异步等待用户输入，所以客户端可以处理来自 WebSocket 服务器的传入消息，同时等待用户输入。

希望的else-clause下面的评论monitor_cmd()解释了我的意思：

import asyncio
import json
import websockets

async def monitor_ws():
    uri = 'ws://localhost:6789'
    async with websockets.connect(uri) as websocket:
        async for message in websocket:
            print(json.dumps(json.loads(message), indent=2, sort_keys=True))

async def monitor_cmd():
    while True:

        sleep_instead = False

        if sleep_instead:
            await asyncio.sleep(1)
            print('Sleeping works fine.')
        else:
            # Seems like I need the equivalent of:
            # line = await asyncio.input('Is this your line? ') …