如何从输入流异步读取？
从输入流异步读取可以通过以下步骤实现：
- 创建一个异步函数，用于处理输入流的读取操作。
- 使用适当的编程语言和库，打开输入流并设置回调函数，以便在数据可用时触发。
- 在回调函数中，读取输入流的数据块，并进行相应的处理。这可以是将数据写入文件、进行数据分析等。
- 如果输入流还有更多数据可用，继续读取并处理，直到没有更多数据为止。
- 在读取完所有数据后，关闭输入流，释放相关资源。
以下是一个示例代码（使用Python和asyncio库）来演示如何从输入流异步读取：
import asyncio
async def read_input_stream(input_stream):
while True:
data = await input_stream.read(1024) # 读取输入流的数据块，每次最多读取1024字节
if not data: # 如果没有更多数据可用，退出循环
break
# 处理读取到的数据，这里只是简单地打印出来
print(data.decode())
input_stream.close() # 关闭输入流
# 使用asyncio库创建一个事件循环
loop = asyncio.get_event_loop()
# 打开输入流，这里使用标准输入作为示例
input_stream = asyncio.StreamReader()
input_protocol = asyncio.StreamReaderProtocol(input_stream)
input_transport, _ = await loop.connect_read_pipe(lambda: input_protocol, asyncio.get_event_loop().child_watcher)
# 调用异步函数来读取输入流
loop.create_task(read_input_stream(input_stream))
# 运行事件循环，直到输入流关闭
loop.run_until_complete(input_transport.wait_closed())这个示例代码使用了Python的asyncio库来实现异步读取输入流的功能。在示例中，我们创建了一个异步函数read_input_stream来处理输入流的读取操作。通过使用asyncio.StreamReader和asyncio.StreamReaderProtocol，我们将标准输入流转换为异步可读取的流，并在事件循环中调用read_input_stream函数来读取数据。
请注意，这只是一个示例，具体的实现方式可能因编程语言和库的不同而有所差异。在实际应用中，您可能需要根据具体情况进行适当的调整和修改。
相关·内容
5回答
coroutine和将来/任务在Python3.5中的区别？
python、python-3.x、python-asyncio
假设我们有一个虚拟函数：
async def foo(arg):
result = await some_remote_call(arg)
return result.upper()
在以下方面有什么区别：
import asyncio
coros = []
for i in range(5):
coros.append(foo(i))
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(coros))
和：
import asyncio
futures = []
for
浏览 14提问于2016-01-12得票数 127
回答已采纳
1回答
调用正常函数中的异步函数
python、python-3.x、asynchronous、python-asyncio
我对python3.6的异步非常陌生
所以问题是，我有一个类，我想在里面加入一些属性。其中一个属性是来自异步函数的返回值。
做这件事的最佳做法是什么？
在init函数中调用event_loop一次以获得返回值？
使__init__函数异步吗？然后在事件循环中运行它？
干杯!
再次更新：
以下是我的代码：
import asyncio
import aioredis
from datetime import datetime
class C:
def __init__(self):
self.a = 1
self.b = 2
浏览 3提问于2018-03-26得票数 13
回答已采纳
1回答
取消主线程不工作的run_in_executor协同线
python、multithreading、python-asyncio
在我的用例中，当同步操作发生时，我向服务进行ping操作。完成同步操作后，我需要停止ping操作，但似乎run_in_executor无法cancel。
import asyncio
import threading
async def running_bg(loop, event):
await loop.run_in_executor(None, running, loop, event)
def running(loop, event):
while True:
if event.is_set():
print("can
浏览 5提问于2022-06-08得票数 0
回答已采纳
1回答
Asyncios等待reader.read()永远等待
python-3.x、sockets、python-asyncio
我有一个小型服务器，编写如下：
async def handle_client(reader, writer):
request = (await reader.read()).decode('utf8') # should read until end of msg
print(request)
response = "thx"
writer.write(response.encode('utf8'))
await writer.drain()
writer.close()
loop = as
浏览 0提问于2020-12-11得票数 3
回答已采纳
1回答
asyncio.get_event_loop()在跟随asyncio.run()时失败
python、python-3.x、python-asyncio
我试图在同步上下文中找到一个调用异步函数的解决方案。
以下是我的参考资料：
但是我发现，asyncio.get_event_loop()在跟踪asyncio.run()时失败了，下面是我的代码来重现这个问题：
import asyncio
async def asyncfunction(n):
print(f'before sleep in asyncfunction({ n })')
await asyncio.sleep(1)
print(f'after sleep in asyncfunction({
浏览 4提问于2021-12-03得票数 0
2回答
异步请求，如何获得速度优势
python、python-asyncio
对Async来说还是个新手，所以很抱歉，现在可以非常准确地表达这个问题了。 我尝试按照示例(https://pythonprogramming.net/asyncio-basics-intermediate-python-tutorial/)使用异步，希望获得多个requests.get()的速度优势，而不是以同步的方式进行。 import asyncio
import time
import requests
async def get_text(url):
print(f"Load {url}")
data = requests.get(url).tex
浏览 30提问于2020-04-04得票数 1
2回答
如何在Python/Tornado中调用带有run_in_executor方法的异步函数？
python、multithreading、tornado
我有一个python文件，其中有一个tornado请求处理程序类，其中包含以下相关代码： executor = concurrent.futures.ThreadPoolExecutor(max_workers = 20)
from tornado.platform.asyncio import AnyThreadEventLoopPolicy
asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())
class MainHandler(tornado.web.RequestHandler):
async def connec
浏览 102提问于2019-05-03得票数 0
回答已采纳
1回答
异步模块是如何工作的，为什么我更新的示例同步运行？
python、async-await、python-asyncio
我已经在Python3.6中为异步尝试了以下代码:示例1：
import asyncio
import time
async def hello():
print('hello')
await asyncio.sleep(1)
print('hello again')
tasks=[hello(),hello()]
loop=asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
产出与预期相符：
hello
hello
hello agai
浏览 4提问于2017-07-26得票数 2
回答已采纳
1回答
Python异步ensure_future
python、python-3.x、python-asyncio
我正在学习Python的asyncio库，我很难理解某些Python函数是如何工作的。
例如：
import asyncio
async def coroutine():
print('Initialize coroutine')
await asyncio.sleep(1)
print('Done !')
loop = asyncio.get_event_loop()
#asyncio.ensure_future(coroutine())
#loop.create_task(coroutine())
loop.run_forever(
浏览 2提问于2018-06-21得票数 0
回答已采纳
1回答
两个异步任务一起工作。一个街区另一个
python、async-await、python-asyncio
我有两个异步任务。
update_x每1秒更新一次x，take_action将检测x和打印x的更改。
如果我删除await asyncio.sleep(1)中的take_action。x的值将永远不会被打印。该程序在take_action中插入while循环。有人能解释一下为什么asyncio.sleep(1)在这里很重要吗？
下面是简化的代码。实际上，update_x()是用来订阅一个url的，它会发布数据。并利用take_action检测数据中的某些值变化。有更好的设计方法吗？
import asyncio
x = 0
def print_x():
print(&#
浏览 3提问于2020-06-24得票数 0
回答已采纳
1回答
使用异步时无法使用os.fork()将多个进程绑定到一个套接字服务器
python、multiprocessing、selector、python-asyncio、epoll
我们都知道，使用异步大大提高了套接字服务器的性能，显然，如果我们能够利用cpu中的所有内核(可能通过多处理模块或os.fork()等)，事情会变得更棒。
现在，我正在尝试构建一个多核套接字服务器演示，一个异步套接字服务器监听每个核心，并且所有绑定到一个端口。只需创建一个异步服务器，然后使用os.fork()，就可以让进程具有竞争力。
然而，当我试图分叉时，单核精细代码会遇到一些麻烦。似乎在epoll选择器模块中注册来自不同进程的相同文件压缩程序存在一些问题。
我给你看了下面的代码，有人能帮我吗？
下面是一个简单的、逻辑上清晰的回显服务器代码，它使用异步：
import os
import a
浏览 1提问于2019-06-01得票数 2
回答已采纳
3回答
如何从同步代码Python调用异步函数
python、asynchronous、python-3.6、fme
所以我被锁定在我的桌面应用程序后面的python 3.6.2解释器上。
我想要的是从同步的方法或函数中调用异步函数。
当从桌面应用程序调用python函数时，它必须是一个不能等待的普通函数。
从桌面应用程序，我能够发送一个url列表，我想要的是从异步事件中的每个url发回响应。
这是我的尝试，我已经标记了SyntaxError，我不知道如何绕过它。
import fmeobjects
import asyncio
import aiohttp
import async_timeout
logger = fmeobjects.FMELogFile()
timeout = 10
class Fe
浏览 20提问于2018-08-09得票数 35
4回答
asyncio.ensure_future对BaseEventLoop.create_task还是简单的协同效应？
python、python-3.x、python-3.5、coroutine、python-asyncio
我已经看过几个基本的Python3.5教程，它们以不同的形式执行相同的操作。在此代码中：
import asyncio
async def doit(i):
print("Start %d" % i)
await asyncio.sleep(3)
print("End %d" % i)
return i
if __name__ == '__main__':
loop = asyncio.get_event_loop()
#futures = [asyncio.ensure_future(do
浏览 10提问于2016-03-31得票数 130
回答已采纳
1回答
Python run_forever()和任务
python、google-cloud-platform、python-asyncio
我修改了这段代码，以便在异步Python: PubSub中使用Google
import asyncio
import datetime
import functools
import os
from google.cloud import pubsub
from google.gax.errors import RetryError
from grpc import StatusCode
async def message_producer():
""" Publish messages which consist of the current date
浏览 2提问于2018-04-18得票数 2
回答已采纳
1回答
如何在ProcessPoolExecutor中使用异步
python、python-3.x、python-asyncio、concurrent.futures
我在网上搜索大量的地址，我想在我的任务中同时使用异步和ProcessPoolExecutor来快速搜索地址。
async def main():
n_jobs = 3
addresses = [list of addresses]
_addresses = list_splitter(data=addresses, n=n_jobs)
with ProcessPoolExecutor(max_workers=n_jobs) as executor:
futures_list = []
浏览 0提问于2019-05-23得票数 4
回答已采纳
1回答
无法使我的异步函数正常工作
python、python-3.x、websocket、jupyter-notebook、python-asyncio
我试图基于WebSocket提供的数据构建一个脚本，但我有一个棘手的问题无法解决。我有两个牢房。
第一项：
msg = ''
stream = {}
async def call_api(msg):
async with websockets.connect('wss://www.bitmex.com/realtime?subscribe=quote:XBTUSD,quote:ETHU19') as websocket:
await websocket.send(msg)
while websocket.open:
浏览 0提问于2019-08-31得票数 1
回答已采纳
1回答
Asyncio任务取消
python、python-asyncio、cancellation
这是我创建的一个测试脚本，用来更好地理解任务取消-
import asyncio
import random
import signal
import traceback
async def shutdown(signame, loop):
print("Shutting down")
tasks = [task for task in asyncio.Task.all_tasks()]
for task in tasks:
task.cancel()
try:
await task
浏览 1提问于2019-05-30得票数 2
2回答
如何在Python < v3.10中的任务中使用asyncio.Condition
python、python-3.x、python-asyncio
为什么我不能在任务中使用asyncio.Condition呢？
c = asyncio.Condition()
async def a():
print("A ..")
# await asyncio.sleep(0.2) # This works
async with c:
# RuntimeError: Task <Task pending coro=<a() running at ..this file..:13>> got Future <Future pending> attached to
浏览 28提问于2022-05-11得票数 0
2回答
等待和异步，协程从未被等待过
python、async-await、coroutine
因此，基本上我正在尝试编写一个程序，它可以通过python脚本控制wiz光。我使用python 3.6是因为其他的限制。这是我用来连接到wizlight的pywizlight项目的链接。
我知道代码说它只能在python 3.7和更高版本中工作，但是我想知道它是否有可能在3.6中工作。我知道如果你没有wiz灯(也许)，可能很难测试它。当我运行它时，我得到了以下错误消息：
RuntimeWarning:
coroutine 'wizlight.turn_on' was never awaited`
light.turn_on(PilotBuilder(brightness
浏览 28提问于2021-02-15得票数 2
回答已采纳
1回答
在什么情况下ensure_future不能真正开始未来？
python、python-asyncio
我试图在Python中异步下载一个文件，在子进程中使用wget。我的代码如下所示：
async def download(url, filename):
wget = await asyncio.create_subprocess_exec(
'wget', url,
'O', filename
)
await wget.wait()
def main(url):
loop = asyncio.get_event_loop()
future = asyncio.ensure_future
浏览 0提问于2018-01-27得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
