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:UsersMartinDesktopfinish', 'w')
    f.close()


utils.do_in_thread(dump_data)

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

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("这就是街舞! 
")
    time.sleep(2)

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

def call():
    print("打电话! 
")
    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 
" %  read_value)
    some_var = read_value + 1
    print("加过后全局变量的值： %d
" %  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 
" %  read_value)
    some_var = read_value + 1
    lock.release()
    print("加过后全局变量的值： %d
" %  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