Python异步库问题

huangapple
go评论94阅读模式
英文:

python async library issue

问题

以下是翻译好的代码部分：

  1. import asyncio
  2. import logging
  4. async def first_function(queue1, queue2):
  5.     while True:
  6.         # do something
  7.         result1 = "some result"
  8.         result2 = {"key": "value"}
  9.         await queue1.put(result1)
  10.         await queue2.put(result2)
  11.         await asyncio.sleep(1)  # 等待1秒
  13. async def second_function(queue1, queue2, queue3):
  14.     while True:
  15.         result1 = await queue1.get()
  16.         result2 = await queue2.get()
  17.         # 对结果执行操作
  18.         logging.info(f"Second function received: {result1}, {result2}")
  19.         await queue3.put(result1)
  20.         await asyncio.sleep(1)  # 等待1秒
  22. async def third_function(queue):
  23.     while True:
  24.         result = await queue.get()
  25.         await asyncio.sleep(40) # 等待40秒
  26.         # 对结果执行操作
  27.         logging.info(f"Third function received: {result}")
  29. async def main():
  30.     queue1 = asyncio.Queue()
  31.     queue2 = asyncio.Queue()
  32.     queue3 = asyncio.Queue()
  33.     task1 = asyncio.create_task(first_function(queue1, queue2))
  34.     task2 = asyncio.create_task(second_function(queue1, queue2, queue3))
  35.     task3 = asyncio.create_task(third_function(queue3))
  36.     await asyncio.gather(task1, task2, task3)

希望这能帮助您解决问题。如果您有其他问题，请随时提出。

英文:

I have the following script, but it work incorecctly. I want to first function always repeat and second function should start if first function return result1, result2. I want to add one more function which should start 40 seconds after second function finished job with result1 from first function. Third function should do something with result2. I want to work it asynchronously using asyncio library because i need the first function to not block for 40 seconds while third function sleep. I tried to use chatgpt, but this tool recommend me wrong things. How to do that?

  1. import asyncio
  2. import logging
  6. async def first_function(queue1, queue2):
  7.     while True:
  8.         # do something
  9.         result1 = "some result"
  10.         result2 = {"key": "value"}
  11.         await queue1.put(result1)
  12.         await queue2.put(result2)
  13.         await asyncio.sleep(1)  # wait for 1 second
  15. async def second_function(queue1, queue2, queue3):
  16.     while True:
  18.            result1 = await queue1.get()
  19.            result2 = await queue2.get()
  20.            # do something with the results
  21.            logging.info(f"Second function received: {result1}, {result2}")
  22.            await queue3.put(result1)
  23.            await asyncio.sleep(1)  # wait for 1 second
  26. async def third_function(queue):
  27.     while True:
  29.            result = await queue.get()
  30.            await asyncio.sleep(40) # wait for 40 seconds
  31.            # do something with the result
  32.            logging.info(f"Third function received: {result}")
  35. async def main():
  36.     queue1 = asyncio.Queue()
  37.     queue2 = asyncio.Queue()
  38.     queue3 = asyncio.Queue()
  39.     task1 = asyncio.create_task(first_function(queue1, queue2))
  40.     task2 = asyncio.create_task(second_function(queue1, queue2, queue3))
  41.     task3 = asyncio.create_task(third_function(queue3))
  42.     await asyncio.gather(task1, task2, task3)

I tried everything i googled and it always block my code for 40 seconds when reach third function

答案1

得分: 0

存在多个问题和反模式：

  • 无需使用asyncio.create_task()，因为当对象是协程时，AsyncIO会自动完成这个操作，
  • 你使用了无界队列，这意味着程序的内存消耗可能会无限增长，
  • third_function 需要40秒，而 first_function 每12秒产生一次元素，这意味着前者无法快速处理元素，
  • 如果 result1 和 result_2 应该同时产生并同时消耗，使用一个队列来存放这两个元素。

你可以使用类似以下的代码：

  1. import asyncio
  3. async def first_function(queue_in: asyncio.Queue):
  4.     while True:
  5.         result1 = "some result"
  6.         result2 = {"key": "value"}
  7.         await queue_in.put((result1, result2))
  9. async def second_function(queue_in: asyncio.Queue, queue_out: asyncio.Queue):
  10.     while True:
  11.         result1, result2 = await queue_in.get()
  12.         print(f"Second function received: {result1}, {result2}")
  13.         await queue_out.put(result1)
  15. async def third_function(queue_out: asyncio.Queue):
  16.     while True:
  17.         result = await queue_out.get()
  18.         await asyncio.sleep(4)
  19.         print(f"Third function received: {result}")
  21. async def main():
  22.     queue_in = asyncio.Queue(maxsize=1)
  23.     queue_out = asyncio.Queue(maxsize=1)
  24.     
  25.     task1 = first_function(queue_in)
  26.     task2 = second_function(queue_in, queue_out)
  27.     task3 = third_function(queue_out)
  28.     
  29.     await asyncio.gather(task1, task2, task3)
  30.     
  31. if __name__ == "__main__":
  32.     asyncio.run(main())

通过指定 maxsize=1，你限制了 first_function 的生产速率，因为现在 third_function 定义了消费速率：只有在 third_function 完成对上一个元素的处理后，first_function（和 second_function）才被允许产生新的元素。

如果你无法控制 first_function 产生元素的速率，你应该实现一种反压策略，例如，如果你无法以足够快的速度处理它们，可以丢弃元素。如果想要在放弃元素之前允许一些等待，可以使用较高的 max size 值。

英文:

There is multiple issues and anti-patterns:

  • there is no need to use asyncio.create_task() since AsyncIO will do that automatically for you when the objects are coroutines,
  • you use unbounded queues, which means that the memory consumption or your program could grow unlimited,
  • third_function takes 40 seconds while first_function produces elements every 12 seconds which means that the former won't have time to process elements fast enough,
  • if result1 and result_2 should be produces at the same time and consumed at the same time, use a single queue to put both elements.

You could use something like this:

  1. import asyncio
  4. async def first_function(queue_in: asyncio.Queue):
  5.     while True:
  6.         result1 = "some result"
  7.         result2 = {"key": "value"}
  8.         await queue_in.put((result1, result2))
  11. async def second_function(queue_in: asyncio.Queue, queue_out: asyncio.Queue):
  12.     while True:
  13.         result1, result2 = await queue_in.get()
  14.         print(f"Second function received: {result1}, {result2}")
  15.         await queue_out.put(result1)
  18. async def third_function(queue_out: asyncio.Queue):
  19.     while True:
  20.         result = await queue_out.get()
  21.         await asyncio.sleep(4)
  22.         print(f"Third function received: {result}")
  25. async def main():
  26.     queue_in = asyncio.Queue(maxsize=1)
  27.     queue_out = asyncio.Queue(maxsize=1)
  28.     
  29.     task1 = first_function(queue_in)
  30.     task2 = second_function(queue_in, queue_out)
  31.     task3 = third_function(queue_out)
  32.     
  33.     await asyncio.gather(task1, task2, task3)
  34.     
  35.     
  36. if __name__ == "__main__":
  37.     asyncio.run(main())

By specifying maxsize=1 you limit the production of first_function because now third_function defines the consumption rate: first_function (and second_function to) is only allowed to produce a new element when third_function finished processing the last one.

If you cannot control the rate at which first_function produces elements, you should implement a back-pressure strategy where for instance you drop elements if you cannot process them fast enough. You can use a higher value for max size if you want to allow some sort of waiting before elements are dropped.

huangapple
  • 本文由 发表于 2023年2月27日 17:22:48
  • 转载请务必保留本文链接：https://go.coder-hub.com/75578655.html
  • async-await
  • asynchronous
  • python
  • python-asyncio
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定