python 多线程

Posted 2021-01-08 山外云

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了python 多线程相关的知识，希望对你有一定的参考价值。

1、进程的线程共享进程的资源

2、线程的目的

异步：我想异步做一件事情，总的执行流继续往下走

并行（多线程）：同时做很多件事情

单线程作用是异步

3、如何把一件事情放到线程中去做

import threading

t = threading.Thread(target=func,args=[])
t.start

只需掌握两个参数

1、target 后面跟做的事情函数名

2、args 后边跟着函数的参数

import threading
import time

def sleep():
    time.sleep(5)
    print(\'5 second\')
    return \'money\'


t = threading.Thread(target=sleep)
t.start()

函数带参数的（参数的小括号可以换成中括号)

原生线程调用方式缺点

1、传参数不得劲

2、函数的返回值无法得到

封装的线程类 utils.py

#coding:gbk

import threading
import time

class FuncThread(threading.Thread):
    
    def __init__(self, func, *args, **kwargs):
        super(FuncThread, self).__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.finished = False
        self.result = None

    def run(self):
        self.result = self.func(*self.args, **self.kwargs)
        self.finished = True

    def is_finished(self):
        return self.finished

    def get_result(self):
        return self.result


def sleep(n):
    time.sleep(n)
    print(\'sleep %s second\' % n)
    return \'money\'


def do_in_thread(func, *args, **kwargs):
    ft = FuncThread(func, *args, **kwargs)
    ft.start()
    return ft

t = do_in_thread(sleep, 2)
time.sleep(2.5)
print(t.get_result())

可以拿到线程的返回结果

封装线程的使用方法

do_in_thread(func_name, 参数)

多线程怎么办？

在for循环中起多个单线程即可

utils.py

#coding:gbk

import threading


class FuncThread(threading.Thread):
    
    def __init__(self, func, *args, **kwargs):
        super(FuncThread, self).__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.finished = False
        self.result = None

    def run(self):
        self.result = self.func(*self.args, **self.kwargs)
        self.finished = True

    def is_finished(self):
        return self.finished

    def get_result(self):
        return self.result


def do_in_thread(func, *args, **kwargs):
    ft = FuncThread(func, *args, **kwargs)
    ft.start()
    return ft

#coding:utf-8

import time
import os
import utils


def check_file_exist(file_name):
    return os.path.exists(file_name)


#超时处理函数

def handle_timeout(func, timeout, *args, **kwargs):
    interval = 1
    
    ret = None
    while timeout > 0:
        begin_time = time.time()
        ret = func(*args, **kwargs)
        if ret:
            break
        time.sleep(interval)
        timeout -= time.time() - begin_time
    
    return ret

def dump_data():
    time.sleep(100)
    f = open(r\'C:\\Users\\Martin\\Desktop\\finish\', \'w\')
    f.close()


utils.do_in_thread(dump_data)

"""在一分钟之内检查桌面上是否有finish这样一个文件，如果有，返回True，
   如果没有，继续检查，知道超时
"""
print(handle_timeout(check_file_exist, 5, r\'C:\\Users\\Martin\\Desktop\\finish\'))

join 方法

作用：使得线程阻塞到函数执行完成之后，主流程才继续往下走

如果程序想要在线程完成后再往下走，那么可以用join完成

参数：默认知道完成后才继续往下走

超时：阻塞的时间

使用方法

ft = do_in_thread(fun_name, 参数)
ft.join()

join使用场景：多线程中

#coding:utf-8

import threading
import time

class FuncThread(threading.Thread):
    
    def __init__(self, func, *args, **kwargs):
        super(FuncThread, self).__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.finished = False
        self.result = None

    def run(self):
        self.result = self.func(*self.args, **self.kwargs)
        self.finished = True

    def is_finished(self):
        return self.finished

    def get_result(self):
        return self.result


def do_in_thread(func, *args, **kwargs):
    ft = FuncThread(func, *args, **kwargs)
    ft.start()
    return ft

def a():
    time.sleep(5)
    print(\'a : done\')

ft = do_in_thread(a)
ft.join()

print(\'测试join\')

加了join方法之后，主执行流等待子线程完成后最后才执行

多线程部署java代码实例

#coding:gbk

import time
import threading


class FuncThread(threading.Thread):
    
    def __init__(self, func, *args, **kwargs):
        super(FuncThread, self).__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.finished = False
        self.result = None

    def run(self):
        self.result = self.func(*self.args, **self.kwargs)
        self.finished = True

    def is_finished(self):
        return self.finished

    def get_result(self):
        return self.result


def do_in_thread(func, *args, **kwargs):
    ft = FuncThread(func, *args, **kwargs)
    ft.start()
    return ft


def handle_timeout(func, timeout, *args, **kwargs):
    interval = 1
    
    ret = None
    while timeout > 0:
        begin_time = time.time()
        ret = func(*args, **kwargs)
        if ret:
            break
        time.sleep(interval)
        timeout -= time.time() - begin_time
    
    return ret

def calc_time(func):
    def _deco(*args, **kwargs):
        begin_time = time.time()
        ret = func(*args, **kwargs)
        cost_time = time.time() - begin_time
        print(\'do %s cost time: %s\' % (func, cost_time))
        return cost_time
    return _deco

#coding:gbk

import time
import utils

def deploy_java():
    print("开始部署java代码")
    time.sleep(5)
    print("部署java代码完成")
    return True

def start_tomcat():
    print("开始启动tomcat")
    time.sleep(5)
    print("启动tomcat完成")
    return True

def health_check():
    print("开始进行java服务健康检查")
    time.sleep(1)
    print("java服务健康检查完成，服务正常")
    return True

def check(thread_obj):
    if not thread_obj.is_finished():
        return
    return thread_obj.get_result()

@utils.calc_time
def deploy(timeout):

    for step in steps:
        thread_obj = utils.do_in_thread(step)
        rst = utils.handle_timeout(check, timeout, thread_obj)
        if not rst:
            print("%s 没有在%s 秒的时间内完成" % (step, timeout))
            print("停止部署任务")
            break

if __name__ == \'__main__\':
    steps = [deploy_java, start_tomcat, health_check]
    print("开始部署")
    #begin_time =  time.time()
    deploy(10)
    #print("部署任务共用了 %s " % str(time.time() - begin_time))

多线程：多个单线程

使用方法：
1）用多线程做一件事情 80%
for i in range(100):
do_in_thread(func_name)
2）多线程做多件事情

#coding:gbk

import time
import threading


class FuncThread(threading.Thread):
    
    def __init__(self, func, *args, **kwargs):
        super(FuncThread, self).__init__()
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.finished = False
        self.result = None
    
    def run(self):
        self.result = self.func(*self.args, **self.kwargs)
        self.finished = True
    
    def get_result(self):
        return self.result
    
    def is_finished(self):
        return self.finished


def do_in_thread(func, *params, **paramMap):
    ft = FuncThread(func, *params, **paramMap)
    ft.start()
    return ft


def calc_time(func):
    def _deco(*args, **kwargs):
        begin_time = time.time()
        ret = func(*args, **kwargs)
        cost_time = time.time() - begin_time
        print(\'do %s cost time: %s\' % (func, cost_time))
        return cost_time
    return _deco


def movie():
    print("这就是街舞! \\n")
    time.sleep(2)

def eating():
    print("吃零食. \\n")
    time.sleep(1)

def call():
    print("打电话! \\n")
    time.sleep(0.5)

@calc_time
def calc_time_without_multi_threading():
    call()
    eating()
    movie()

@calc_time
def calc_time_with_multi_threading():

    func_names = [movie, eating, call]

    thread_objs = []
    for func_name in func_names:
        t = do_in_thread(func_name)
        thread_objs.append(t)

    for thread_obj in thread_objs:
        thread_obj.join()

#print(\'不使用线程占用时间：%s\' % calc_time_without_multi_threading())
print(\'使用线程占用时间：%s\' % calc_time_with_multi_threading())

方式：
1）通过一个列表在for循环中收集所有线程的对象
2）遍历列表，执行每个对象join

结论：
多线程全部join占用总时间是：多线程中执行最长的那个线程的时间

多线程安全问题----锁

from threading import Lock

lock = Lock()
lock.acquire()
lock.release()

1）初始化锁对象
lock = Lock()
2）获取锁
lock.acquire()
3）释放锁
lock.release()

#coding:utf-8

import threading
from threading import Lock


some_var = 0


class FuncThread(threading.Thread):
    
    def __init__(self, func, *params, **kwparams):
        threading.Thread.__init__(self)
        self.func = func
        self.params = params
        self.kwparams = kwparams
        self.result = None
        self.finished = False
    
    def run(self):
        self.result = self.func(*self.params, **self.kwparams)
        self.finished = True
    
    def get_result(self):
        return self.result
    
    def is_finished(self):
        return  self.finished

def do_in_thread(func, *params, **paramMap):
    ft = FuncThread(func, *params, **paramMap)
    ft.start()
    return ft


def add_some_var():
    global some_var
    read_value = some_var
    print("当前全局变量的值： %d \\n" %  read_value)
    some_var = read_value + 1
    print("加过后全局变量的值： %d\\n" %  some_var)


def do_add_in_thread():
    threads = []
    for i in range(500):
        t = do_in_thread(add_some_var)
        threads.append(t)
    for t in threads:
        t.join()
    print("预期some_var为500")
    print("实际结果是 %d" % (some_var,))


do_add_in_thread()

加锁后

#coding:utf-8

import threading
from threading import Lock


some_var = 0
lock = Lock()

class FuncThread(threading.Thread):
    
    def __init__(self, func, *params, **kwparams):
        threading.Thread.__init__(self)
        self.func = func
        self.params = params
        self.kwparams = kwparams
        self.result = None
        self.finished = False
    
    def run(self):
        self.result = self.func(*self.params, **self.kwparams)
        self.finished = True
    
    def get_result(self):
        return self.result
    
    def is_finished(self):
        return  self.finished

def do_in_thread(func, *params, **paramMap):
    ft = FuncThread(func, *params, **paramMap)
    ft.start()
    return ft


def add_some_var():
    global some_var
    lock.acquire()
    read_value = some_var
    print("当前全局变量的值： %d \\n" %  read_value)
    some_var = read_value + 1
    lock.release()
    print("加过后全局变量的值： %d\\n" %  some_var)

def do_add_in_thread():
    threads = []
    for i in range(500):
        t = do_in_thread(add_some_var)
        threads.append(t)
    for t in threads:
        t.join()
    print("预期some_var为500")
    print("实际结果是 %d" % (some_var,))

do_add_in_thread()

线程池原理

安装futures模块（tornado 使用线程池 futures)

使用：
from concurrent.futures import ThreadPoolExecutor
t = ThreadPoolExecutor(num)
t.submit(func_name)

t = ThreadPoolExecutor(10)
t.submit(do_something）

# coding:utf8

import time 
from concurrent.futures import ThreadPoolExecutor

TEST_COUNT = 100
THREADS_SUM = 10
thread_pool = ThreadPoolExecutor(THREADS_SUM)


def test(num):
    time.sleep(1)
    print(\'中秋快乐;\')


for i in range(TEST_COUNT):
    thread_pool.submit(test,i)

python 执行 shell指令

def create_process(cmd):
    p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    result = p.stdout.read()
    code = p.wait()
    return code, result

以上是关于python 多线程的主要内容，如果未能解决你的问题，请参考以下文章

[Python3] 043 多线程简介

python中的多线程和多进程编程

多线程 Thread 线程同步 synchronized

多个用户访问同一段代码

在 Python 多处理进程中运行较慢的 OpenCV 代码片段

线程学习知识点总结