python異步任務隊列示例
更新時間:2014年04月01日 14:42:36 作者:
這篇文章主要介紹了python異步任務隊列示例,需要的朋友可以參考下
很多場景為了不阻塞,都需要異步回調機制。這是一個簡單的例子,大家參考使用吧
復制代碼 代碼如下:
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import logging
import queue
import threading
def func_a(a, b):
return a + b
def func_b():
pass
def func_c(a, b, c):
return a, b, c
# 異步任務隊列
_task_queue = queue.Queue()
def async_call(function, callback, *args, **kwargs):
_task_queue.put({
'function': function,
'callback': callback,
'args': args,
'kwargs': kwargs
})
def _task_queue_consumer():
"""
異步任務隊列消費者
"""
while True:
try:
task = _task_queue.get()
function = task.get('function')
callback = task.get('callback')
args = task.get('args')
kwargs = task.get('kwargs')
try:
if callback:
callback(function(*args, **kwargs))
except Exception as ex:
if callback:
callback(ex)
finally:
_task_queue.task_done()
except Exception as ex:
logging.warning(ex)
def handle_result(result):
print(type(result), result)
if __name__ == '__main__':
t = threading.Thread(target=_task_queue_consumer)
t.daemon = True
t.start()
async_call(func_a, handle_result, 1, 2)
async_call(func_b, handle_result)
async_call(func_c, handle_result, 1, 2, 3)
async_call(func_c, handle_result, 1, 2, 3, 4)
_task_queue.join()
相關文章
用?Python?腳本實現電腦喚醒后自動拍照并截屏發(fā)郵件通知
這篇文章主要介紹了用?Python?腳本實現電腦喚醒后自動拍照并截屏發(fā)郵件通知,文中詳細的介紹了代碼示例,具有一定的 參考價值,感興趣的可以了解一下2023-03-03
Python使用Selenium時遇到網頁<body>劃不動的問題解決方法
如果在使用 Selenium 時遇到網頁的 <body> 劃不動的問題,這通常是因為頁面的滾動機制(例如,可能使用了一個具有固定高度的容器或自定義的滾動條)導致無法通過簡單的 JavaScript 實現滾動,可以通過以下方法來解決該問題2024-10-10

