标签: python-asyncio

所以我有一个倒计时脚本，它看起来像这样：

import time, threading, asyncio
def countdown(n, m):
    print("timer start")
    time.sleep(n)
    print("timer stop")
    yield coro1

async def coro1():
    print("coroutine called")

async def coromain():
    print("first")
    t1 = threading.Thread(target=countdown, args=(5, 0))
    t1.start()
    print("second")

loop = asyncio.get_event_loop()
loop.run_until_complete(coromain())
loop.stop()

我想要它做的很简单：

Run coromain
Print "first"
Start thread t1, print "timer start" and have it wait for 5 seconds
In the mean time, print "second"
after 5 seconds, print "timer stop"
exit

但是，当我运行此代码时，它会输出：

Run coromain
Print "first"
Print "second"
exit

我很困惑为什么它会这样做。谁能解释我在这里做错了什么？

我对图书馆越来越熟悉，但我被以下情况难住了：

我希望不断处理来自两个不同网站的 websocket 的更新消息。

但是我无法使用单个会话变量实现这一点，因为 aiohttp.ClientSession() 对象应该存在于协程中。

import asyncio
import aiohttp

url1 = 'wss://example.com'

async def main():
    session = aiohttp.ClientSession()
    async with session.ws_connect(url1) as ws1:
        async for msg in ws1:
            # do some processing

loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()

以上将适用于单个 websocket 连接。但是因为ws:中的async for msg是一个无限循环，所以我看不到我可以把这个异步循环的 ws2 版本放在哪里。

我需要 Python 网络框架工作Quart 的帮助，更具体地说是 websockets。我希望能够在连接时注册客户端（将其添加到 python 列表），并在断开连接时取消注册（从 python 列表中删除它）。我在网上能找到的最接近的是这段代码：

connected = set()

async def handler(websocket, path):
    global connected
    # Register.
    connected.add(websocket)
    try:
        # Implement logic here.
        await asyncio.wait([ws.send("Hello!") for ws in connected])
        await asyncio.sleep(10)
    finally:
        # Unregister.
        connected.remove(websocket)

来源

但这不适用于夸脱 websockets。

帮助将不胜感激。

我尝试在 Tornado 和 Redis 上构建一个具有两个 API 端点的简单系统：

1. 从 Redis 读取值的 API，或等待该值存在（使用BRPOP: value = yield from redis.brpop("test")）
2. 写入此值的 API（带有LPUSH: redis.lpush("test", "the value")）。

所以我希望能够以任何顺序调用这些 API。事实上，如果我调用 2. 然后 1.，它会按预期工作，对 1. 的调用立即返回值。

问题是，如果我调用 1. 然后 2.，两个请求都会阻塞，永远不会返回。

同时，当请求阻塞时，我仍然可以LPUSH/BRPOP直接在 Redis 中，甚至可以使用相同的键。同样，我可以在 Tornado 中调用其他处理程序。所以我猜这个块既不在 Redis 也不在 Tornado 中，而是在我对 aioredis 的使用中？也许是异步循环？但我不明白我错在哪里。任何提示？

谢谢你的帮助。

这是我的代码：

import tornado.ioloop
import tornado.web
from tornado import web, gen
from tornado.options import options, define
import aioredis
import asyncio


class WaitValueHandler(tornado.web.RequestHandler):
    @asyncio.coroutine
    def get(self):
        redis = …

假设我有一个类，并且想从磁盘并行读取几个文件，并对类参数进行参数化。什么是最正确的方法（以及如何）？

• 主线程应等待 load_data() 操作结束，然后再发生其他任何事情。

我考虑过线程，因为它只是 I/O 操作。

非并行实现示例（1-Threading）：

import pandas as pd


class DataManager(object):
    def __init__(self):
        self.a = None
        self.b = None
        self.c = None
        self.d = None
        self.e = None
        self.f = None

    def load_data(self):
        self.a = pd.read_csv('a.csv')
        self.b = pd.read_csv('b.csv')
        self.c = pd.read_csv('c.csv')
        self.d = pd.read_csv('d.csv')
        self.e = pd.read_csv('e.csv')
        self.f = pd.read_csv('f.csv')

if __name__ == '__main__':
    dm = DataManager()
    dm.load_data()
    # Main thread is waiting for load_data to finish.
    print("finished loading data")

我需要创建一个软件，它同时从 web 套接字和管道接收消息，并在另一个通道上发送消息（它从套接字接收，创建一个新线程并发送到管道。以同样的方式从管道接收，创建一个新线程并发送到套接字）。

我有一个多线程问题，在程序启动时我必须启动方法socket_receiver，pipe_receiver但我只能启动pipe_receiver. 我试图消除所有的代码，并只保留socket_receiver和pipe_receiver，但它只是在进入while True的pipe_receiver。

import asyncio
import sys
import json
from concurrent.futures.thread import ThreadPoolExecutor
import websockets

# make the Pool of workers
executor = ThreadPoolExecutor(max_workers=10)
# Make connection to socket and pipe
header = {"Authorization": r"Basic XXXX="}
connection = websockets.connect('wss://XXXXXXXX', extra_headers=header)


async def socket_receiver():
    """Listening from web socket"""
    async with connection as web_socket:
        while True:
            message = await web_socket.recv()
            # send the message to …

是否可以使用 asyncio 为每个函数以不同的间隔定期运行 2 个不同的函数f？

如果是 - 您能否提供代码示例。

import asyncio

async def f1(host,*arg):
    # call every 1 sec
    pass

async def f2(url,*args):
    # call every 2 sec
    pass

预期输出：

00:00 f1 for 1.1.1.1

00:01 f1 for 1.1.1.1

00:02 f2 for 'google.com'

00:02 f1 for 1.1.1.1

00:03 f1 for 1.1.1.1

00:04 f2 for 'google.com'

我想通过使用库 python-telegram-bot 和 discord.py （版本 1.0.0）来制作一个在不和谐和电报之间进行通信的机器人。然而问题是 discord.py 使用异步函数和 python-telegram-bot 线程。使用下面的代码，对于在 discord 中发布的消息（机器人将它们正确发送到电报），一切正常，但反之则行不通（机器人从电报获取消息并将其发送到不和谐）。我以前在尝试channel.send在同步函数中运行 discords函数时遇到了语法/运行时错误的问题（因此要么只返回一个生成器对象，要么抱怨我无法使用await在同步函数中使用）。然而，与此同时，python-telegram-bot 的MessageHandler需要一个同步函数，所以当给他一个异步函数时，Python 抱怨从来没有为异步函数调用“await”。我现在试图用async_to_sync方法从asgiref库运行我的异步broadcastMsg从MessageHandler，但是代码仍然不将消息发送给不和谐！它似乎正确地调用了该函数，但仅在 line 之前print('I get to here')。没有错误显示，也没有不和谐的消息弹出。我想这与我必须在 discord.py 事件循环中将该函数注册为任务这一事实有关，但是注册仅在它发生之前才有效botDiscord.run(TOKENDISCORD)已经被执行，这当然必须在之前发生。因此，将我的问题归结为一个问题：我如何能够与来自另一个线程（来自电报MessageHandler）的 discord.py 事件循环进行交互。或者，如果这是不可能的：如何在不处于 discord.py 事件循环中的情况下使用 discord.py 发送消息？

感谢您的帮助

import asyncio
from asgiref.sync import async_to_sync
from telegram import Message as TMessage
from telegram.ext import (Updater,Filters,MessageHandler)
from discord.ext import commands
import discord

TChannelID = 'someTelegramChannelID'
DChannel = 'someDiscordChannelObject'

#%% define functions / commands
prefix = …