python多线程加锁异步处理装饰器

摘要: python

前言：

虽谈python多线程带有全局锁PIL，似乎对性能提升没什么意义，一般考虑多进程或者协程，但PIL没有被去掉还是应该有它的用意的，一来不用考虑线程安全问题，在者处理小事务，多线程还是有一定程序上的便利性的，以下将简单描述一下应用业务上的多线程技术。

问题描述：

1.对于web请求，需要某些api需要做异步处理，例如：对于一个web请求，数据校验通过了，然后进行保存，对于用户来说，需要得到一个快速的响应才是很好的体验，检验成功的数据，也没必要让用户作无谓的保存等待。

2.对于保存的数据，需要异常的抛出逻辑异常让上层捕捉。

3.对于异步处理，需要加入队列（加锁），防止数据保存混乱。

4.开发使用需要简单，只需要加入装饰器即可。

5.被装饰的函数，有独立的锁，互不影响浪费时间。

见代码：

# encoding:utf-8
# usr/bin/python
import time
import hashlib
import pickle
import threading
from functools import wraps

lock_dict = {}

class MyThread(threading.Thread):
def __init__(self, func, args=None, kwargs=None):
super(MyThread, self).__init__()
self.func = func
self.args = args
self.kwargs = kwargs
self.key = hashlib.sha1(pickle.dumps((self.func.func_name,))).hexdigest()
self.result = None

@property
def lock(self):
lock = lock_dict.get(self.key)
if not lock:
lock = threading.RLock()
lock_dict[self.key] = lock
return lock

def run(self):
""""重写start方法,加锁,逻辑处理异常抛出异常,这里可以拓展结果返回"""
try:
if self.lock.acquire():
self.result = self.func(*self.args)
except:
raise
finally:
self.lock.release()

def get_result(self):
try:
return self.result  # 获取时候需要t.join(), 不然主线程比子线程跑的快,会拿不到结果
except:
raise

def thread_function(function):
"""多线程路由函数独立锁（函数名作为唯一识别,保证函数线程安全）"""

@wraps(function)
def run(*args, **kwargs):
try:
t = MyThread(function, args=args, kwargs=kwargs)
t.start()
except:
raise

return run

def thread_lock(function):
"""多线程路由函数独立锁（函数名作为唯一识别,保证函数线程安全）"""

@wraps(function)
def run(*args, **kwargs):
try:
key = hashlib.sha1(pickle.dumps((function.func_name,))).hexdigest()
lock = lock_dict.get(key)
if not lock:
lock = threading.RLock()
lock_dict[key] = lock
if lock.acquire():
function(*args, **kwargs)
except:
raise
finally:
lock.release()

return run

a = 0

@thread_function
def test(j, k=1):
global a
time.sleep(5)
a += 1
print "a{},j{},k{}".format(a, j, k)

b = 0

@thread_lock
def test2(j, k=1):
global b
time.sleep(5)
b += 1
print "b{},j{},k{}".format(b, j, k)

def run_1():
for ca in range(10):
print "a"
test(1, 2)

def run2():
for ca in range(10):
print "b"
t = threading.Thread(target=test2, args=(1,), kwargs={"k": 2})  # 异步处理数据避免阻塞
t.start()

if __name__ == "__main__":
run_1()
run2()

输出：

a
a
a
a
a
a
a
a
a
a
b
b
b
b
b
b
b
b
b
b
a1,j1,k2
b1,j1,k2
a2,j1,k2
b2,j1,k2
a3,j1,k2
b3,j1,k2
a4,j1,k2
b4,j1,k2
a5,j1,k2
b5,j1,k2
a6,j1,k2
b6,j1,k2
a7,j1,k2
b7,j1,k2
a8,j1,k2
b8,j1,k2
a9,j1,k2
b9,j1,k2
a10,j1,k2
b10,j1,k2