Python_Example_Thread 线程 学习/经验/示例

 Author: 楚格

2018-11-17     17:34:58

IDE： Pycharm2018.02   Python 3.7   

KeyWord :  线程 threading Thread

Explain:  

-----------------------------------------------------

--

# coding=utf-8
#---------------------------------
'''
# Author  : chu ge 
# Function: 线程 thread
#
'''
#---------------------------------
'''
# --------------------------------
# 导入模块 
# 1.系统库
# 2.第三方库
# 3.相关定义库
# --------------------------------
'''
# 1.系统库
import sys
import os
import time
import random

#2.第三方库

#进程导入模块
# from multiprocessing import Process
from multiprocessing import Pool
# from multiprocessing import Queue
# from multiprocessing import Manager

#进程导入模块
import threading
from threading import Thread
from threading import Lock
from queue import Queue






#
'''
============================================================================
#》》》》》》》》》》

线程

-----------------------
python的thread模块比较底层，
threading模块做了些包装，可以更加方便的被使用

进程 VS 线程
功能的不同
进程，能够完成多任务，如一台电脑上可以同时运行多个微信
线程，能够完成多任务，如一个微信界面中可以运行多个窗口
定义的不同
进程，是系统进行资源分配和调度的一个独立单位。
线程，是进程的一个实体，是CPU调度和分派的基本单位，它是比进程更小的
能独立运行的基本单位。
    线程自己基本不拥有系统资源，只拥有一点在运行中必不可少的资源，
    如：程序计数器，一组寄存器和栈，但是它可与同属一个进程的
    其他线程共享进程所拥有的全部的资源。

区别
一个程序至少有一个进程，一个进程至少有一个线程。
线程的划分尺度小于进程，资源比进程少，使得多线程程序的并发性高。
进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大提高程序的运行效率
线程不能够独立执行，必须依存在进程中
优缺点
线程和进程在使用上各有优缺点：线程执行开销小但不利于资源管理和保护
而进程正相反。

----------------------------------------------
1. 线程 使用threading 模块

-----------------------
1.1 单 / 多线程执行
e.g:
#1如果多个线程执行同一函数，各是各的，互不影响
def Function_Say_Sorry():
    print("A say: Sorry !")
    time.sleep(1)
if __name__ == "__main__":
    # 单线程
    for var in range(10):
        Function_Say_Sorry()

    # 多线程执行
    for var in range(10):
        # 创建线程
        thread_name = threading.Thread(target = Function_Say_Sorry)
        thread_name.start()
result:
1)可以看出使用了多线程并发操作，花费时间要短很多
2)创建好的线程，需要调用start()方法来启动

-----------------------
e.g:

if __name__ == "__main__":

result:

-----------------------
1.2 主线程会等待所有的子线程结束后才结束
e.g:
def Function_Sing():
    for var in range(10):
        print("正在唱歌 %d" % var)
        time.sleep(1)

def Function_Dance():
    for var in range(5):
        print("正在跳舞 %d" % var)
        time.sleep(1)
        
if __name__ == "__main__":
        # 主线程等待子线程结束而结束
    print("--- start ---:
 %s" % (time.time()))

    time_A = threading.Thread(target = Function_Sing)
    time_B = threading.Thread(target = Function_Dance)
    time_A.start()
    time_B.start()
    time.sleep(5) #
    print("--- finish ---:
%s" % (time.time()))
    # 查看线程数量
    while True:
        length = len(threading.enumerate())
        print("当前运行的线程数为： [%d] " % (length))
        if length <= 1 :
            break
        time.sleep(0.5)
result:

-----------------------
1.3 threading 
1.3.1 线程执行代码的封装
通过使用threading模块能完成多任务的程序开发，为了每个线程更完美
所有使用threading模块时，往往定义新的子类class,
只要继承threading.Thread就可以了，然后重写run方法

e.g:
#线程执行代码的封装
class Class_MyThread(threading.Thread):
    def run(self):
        for var in range(10):
            time.sleep(1)
            message = "I am " + self.name + " @ " + str(var)
            print("message: %s" % (message))


if __name__ == "__main__":
    thread_name = Class_MyThread()
    thread_name.start()
    
result:
python的threading.Thread类有一个run方法，
用于定义线程的功能函数，可以在自己的线程类中覆盖还方法。
而创建自己线程实例后，通过thread类start方法，可以启动该线程，
交给python虚拟机进行调度，当该线程获得执行的机会时，就会调用run方法执行线程。

-----------------------
1.3.2 线程的执行顺序
e.g:
# 线程的执行顺序
  #重复封装类
def Function_Test_A():
    for var in range(10):
        thread_name_local = Class_MyThread()
        thread_name_local.start()
if __name__ == "__main__":
    Function_Test_A()
    
result:
从执行结果可以看出，多线程程序的执行顺序是不确定的
当执行到sleep语句时，线程将会被阻塞(Blocked),到sleep结束后，
线程进入就绪(Runnable)状态，等待调度，而线程调度就讲自行选择一个线程执行。
上面代码可以保证运行完成的run,但线程启动顺序和run函数中每次循环的执行顺序都不确定。

总结
1） 每个线程一定会有一个name，尽管示例中没有指定线程对象的name，
但是python会自动为线程指定一个名字。
2)当线程的run函数结束时，该线程完成
3)无法控制线程调度程序，但是可以通过别的方式来影响调度的方式
4)线程的几种状态

     启动         调度        结束
新建 -----> 就绪 <-----> 运行 -----> 死亡
                        /
         满足条件       等待条件
                     /
               等待(阻塞) 


-----------------------
1.4.1 多线程-共享全局变量

e.g:
def Function_Work_A():
    global global_var_number_A

    for var in range(10):
        global_var_number_A += 1
    print("work A ,number is %d" % (global_var_number_A))


def Function_Work_B():
    global global_var_number_A

    print("work B ,number is %d" % (global_var_number_A))

if __name__ == "__main__":
    global_var_number_A = 100

    print("创建线程之前，number is %d" %(global_var_number_A))
    thread_name_A = threading.Thread(target = Function_Work_A )
    thread_name_A.start()
    time.sleep(1)

    thread_name_B = threading.Thread(target = Function_Work_B )
    thread_name_B.start()

result:
    创建线程之前，number is 100
    work A ,number is 110
    work B ,number is 110

-----------------------
1.4.2 列表当做实参传递到线程中
e.g:
# 列表当做实参传递到线程中
def Function_Work_C(nums):
    local_var_number = nums

    local_var_number.append("CC")
    print("work C ,number is %s" % (local_var_number))


def Function_Work_D(nums):
    local_var_number = nums
    time.sleep(1)
    print("work D ,number is %s" % (local_var_number))

if __name__ == "__main__":
    global_nums = [11,22,33,44,55]
    thread_name_C = threading.Thread(target=Function_Work_C, args=(global_nums,))
    thread_name_C.start()
    thread_name_D = threading.Thread(target=Function_Work_D, args=(global_nums,))
    thread_name_D.start()

result:
    work C ,number is [11, 22, 33, 44, 55, 'CC']
    work D ,number is [11, 22, 33, 44, 55, 'CC']
总结：
在一线程内的所有线程共享全局变量，能够在不适用其他方式的前提下，
完成多线程之间的数据共享，这点要比多进程要好。
缺点就是，线程是对全局变量随意遂改可能造成多线程之间，
对全局变量的混乱，即线程非安全。

-----------------------
1.5 线程不安全  同步

e.g:
# 同步
def Function_Test_B():
    global global_var_number
    for var in range(1000000):
        global_var_number += 1
    print("Test B ,number is %d" % (global_var_number))

def Function_Test_C():
    global global_var_number
    for var in range(1000000):
        global_var_number += 1
    print("Test C ,number is %d" % (global_var_number))

if __name__ == "__main__":
    global_var_number = 0
    thread_name_E = threading.Thread(target = Function_Test_B)
    thread_name_E.start()
    time.sleep(3)
    thread_name_F = threading.Thread(target = Function_Test_C)
    thread_name_F.start()

    print("number: %s" % (global_var_number))

result:
    Test B ,number is 1000000
    number: 1059644
    Test C ,number is 2000000

同步：就是协同步调，按照预定的先后次序进行运行。
同指协同，协助，相互配合
如，进程/线程同步，可以理解为进程/线程A和B一块配合，A执行到一定程度时
要依靠B的某个结果，于是停下来，示意B运行，在将结果给A，A继续执行操作。

解决问题：线程同步
1.系统调用thread_name_E,然后获取到num的值为0，
此时上一把锁，即不允许其他现在操作num
2.对num的值进行+1
3.解锁，此时num的值为1，其他的线程就可以使用num了，
而且num的值是0而不是1
4.同理，其他线程在对num进行修改时，都要先上锁，处理完成后再解锁。
在上锁的整个过程中不允许其他线程访问，就保证了数据的正确性。

-----------------------
1.6 互斥锁

当多个线程几乎同时修改某个共享数据的时候，需要进行同步控制。
线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。

互斥锁为资源引入一个状态： 锁定/非锁定

某个线程要更改共享数据时，先将其锁定，此时资源的状态为锁定，
其他线程不能更改，直到该线程释放资源，将资源的状态变成非锁定，
其他的线程才能再次锁定该资源。
互斥锁保证每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。


threading模块定义Lock类，方便处理锁定：
#创建锁
mutex = threading.Lock()
#锁定
mutex.acquire([blocking])
#释放
mutex.release()

在锁定方法中，acquire可以有个blocking参数
如果设定blocking为True，则当前线程会阻塞，直到获得这个锁为止，如果没有指定，那么默认为True
如果设定blocking为Fasle,则当前线程不会阻塞。


上锁解锁过程：
当一个线程调用锁的acpuire()方法获得解锁时，锁就进入Locked状态。
每次只有一个线程可以获得锁。如果此时另外一个线程试图获得这个锁，
该线程就会变为blocked状态，称为阻塞，直到拥有锁的线程调用锁release()方法释放后，锁进入unlocked状态
线程调度程序从处于同步阻塞状态的线程中选择一个获得锁，并使得该线程进入运行(running)

总结：
锁的好处：
确保了某段关键代码只能由一个线程从头到尾完整的执行
锁的坏处：
阻止了多线程并发执行，包含锁的某段代码实际上只能以单线程模式执行，效率降低了。
由于可以存在多个锁，不同的线程持有不同的锁，并试图获得对方持有的锁，可能造成死锁。


e.g:
# 互斥锁
def Function_Test_D():
    global global_var_number
    # Test_D和Test_F线程都在抢着上锁，对这个锁，如果一方上锁成功，
    # 导致另一方会堵塞（一直等待）到这个锁被解锁为止
    mutexFlag = mutex.acquire(True)
    for var in range(1000000):
        # True 表示堵塞，如果这个锁在上锁之前已经被上锁了，那么这个线程会在
        #False 表示非堵塞，即不管本次调用能够成功上锁，都会卡在这，而继续
        if mutexFlag:
            global_var_number += 1
    # 用来对mutex指向的这个锁，进行解锁，只要解锁了，那么接下来会让所有
    # 因为这个锁，被上了锁而堵塞的线程进行抢着上锁
    mutex.release()

    print("Test D ,number is %d" % (global_var_number))

def Function_Test_E():
    global global_var_number

    for var in range(1000000):
        mutexFlag = mutex.acquire(True)
        if mutexFlag:
            global_var_number += 1
        mutex.release()

    print("Test E ,number is %d" % (global_var_number))

def Function_Test_F():
    global global_var_number
    mutexFlag = mutex.acquire(True)
    for var in range(1000000):
        if mutexFlag:
            global_var_number += 1
    mutex.release()

    print("Test F ,number is %d" % (global_var_number))

if __name__ == "__main__":
    global_var_number = 0
    # 创建锁
    mutex = threading.Lock()

    thread_name_D = threading.Thread(target=Function_Test_D)
    thread_name_D.start()

    thread_name_E = threading.Thread(target = Function_Test_E)
    thread_name_E.start()

    thread_name_F = threading.Thread(target = Function_Test_F)
    thread_name_F.start()

    print("number: %s" % (global_var_number))
    
result:

-----------------------
1.7 多线程-非共享数据

对于全局变量，在多线程中要格外小心，否则容易造成数据错乱的情况发生
在多线程开发中，全局变量是多个线程都是共享的数据，而局部变量等是各自线程，是非共享的。

e.g:
class Class_My_Thread(threading.Thread):
    # 重写 构造方法
    def __init__(self,number,SleepTime):
        threading.Thread.__init__(self)
        self.num       = number
        self.SleepTime = SleepTime

    def run(self):
        self.num += 1  # 局部变量
        time.sleep(self.SleepTime)
        print("线程 (%s), number = (%s)" % (self.name,self.num))

if __name__ == "__main__":
    # 创建锁
    mutex = threading.Lock()
    thread_name_G = Class_My_Thread(100, 3)
    thread_name_G.start()

    thread_name_H = Class_My_Thread(200, 1)
    thread_name_H.start()

    thread_name_I = Class_My_Thread(300, 2)
    thread_name_I.start()
    
result:
    线程 (Thread-2), number = (201)
    线程 (Thread-3), number = (301)
    线程 (Thread-1), number = (101)
局部变量是不共享数据的。

-----------------------
1.8  死锁

在线程共享多个资源的时候，如果二个线程分别占有一部分资源
并且同时等待对方的资源，就会造成死锁。
尽管死锁很少发生，但一旦发生就会造成应用的停止响应。

e.g:
# 死锁
class Class_My_Thread_A(threading.Thread):
    def run(self):
        if mutexA.acquire():
            print(self.name + "--do 1 up --")
            time.sleep(1)

            if mutexB.acquire():
                print(self.name + "--do 1 down --")
                mutexB.release()
            mutexA.release()


class Class_My_Thread_B(threading.Thread):
    def run(self):
        if mutexB.acquire():
            print(self.name + "--do 2 up --")
            time.sleep(1)

            if mutexA.acquire():
                print(self.name + "--do 2 down --")
                mutexA.release()
            mutexB.release()

if __name__ == "__main__":
    # 死锁  不能执行的效果
    mutexA = threading.Lock()
    mutexB = threading.Lock()
    thread_name_J = Class_My_Thread_A()
    thread_name_J.start()
    thread_name_K = Class_My_Thread_B()
    thread_name_K.start()

    print("...")
    
result:
    #避免死锁
    # 程序设计时，要尽量避免，银行家算法
    # 添加超时时间等 acquire（timeout）很重要！！！
    
    
-----------------------
1.9.1 同步应用
多线程有序执行

可以使用互斥锁完成多个任务，有序的进程工作，这就是线程的同步

e.g:
# 多线程有序执行
class Class_Task_C(threading.Thread):
    def run(self):
        while True:
            if mutexC.acquire():
                print("--- Task C ---[%s]" % (self.name))
                time.sleep(1)
                mutexD.release()

class Class_Task_D(threading.Thread):
    def run(self):
        while True:
            if mutexD.acquire():
                print("--- Task D ---[%s]" % (self.name))
                time.sleep(1)
                mutexE.release()

class Class_Task_E(threading.Thread):
    def run(self):
        while True:
            if mutexE.acquire():
                print("--- Task F ---[%s]" % (self.name))
                time.sleep(1)
                mutexC.release()
                
if __name__ == "__main__":
    # 同步应用
    # 创建锁
    mutexC = threading.Lock()

    mutexD = threading.Lock()
    mutexD.acquire()
    mutexE = threading.Lock()
    mutexE.acquire()

    thread_name_L = Class_Task_C()
    thread_name_L.start()

    thread_name_M = Class_Task_D()
    thread_name_M.start()

    thread_name_N = Class_Task_E()
    thread_name_N.start()
result:
    --- Task C ---[Thread-1]
    --- Task D ---[Thread-2]
    --- Task F ---[Thread-3]
    --- Task C ---[Thread-1]
    --- Task D ---[Thread-2]
    --- Task F ---[Thread-3]
    --- Task C ---[Thread-1]
    --- Task D ---[Thread-2]

-----------------------
2.0 Queue 队列

Queue使用说明
1) 对于Queue，在多线程通信之间扮演重要的角色
2) 添加数据队列中，使用put()方法
3) 从队列中获取数据，使用get()方法
4) 判断队列中是否还有数据，使用qsize()方法

生产者消费者模式说明：
在线程世界里。生成者就是生产数据的线程，消费者就是消费数据的线程。
在多线程开发中，如果生产者处理速度很快，而消费者处理速度很慢，
那么生成者就必须等待消费者处理完，才能继续生产数据。同样道理
如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。
为了解决这个问题引入了生产者和消费者模式。

生产者消费者模式，是通过一个容器来解决生产者和消费者的强耦合问题。
生产者和消费者之间彼此之间不直接通讯，而通过阻塞队列来进行通讯，
所有生产者生产数据之后不用等待消费者处理，直接扔给阻塞队列，
消费者不找生产者要数据，而是直接从阻塞队列里取出数据，
阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
这个阻塞队列就是用来给生产者和消费者解耦的。
纵观大多数设计模式，都会找到一个第三者出来进行解耦。


e.g:
class Class_Producer_F(threading.Thread):
    def run(self):
        # global queue
        var_count = 0

        while True:
            if queue_name.qsize() < 1000 :
                # 每次就生产100个商品，二个线程做个事
                for var in range(100):
                    var_count += 1
                    message = "生成产品" + str(var_count)
                    queue_name.put(message)
                    # print("-- Producer F --[%s]" % (self.name))
            time.sleep(0.5)

class Class_Consumer_G(threading.Thread):
    def run(self):
        # global queue

        while True:
            if queue_name.qsize() > 100:
                # 若队列数量大于100，每个线程就取出3个商品，五个线程做这个事
                for var in range(3):
                    message = self.name + "消费了" + queue_name.get()
                    print("message: (%s)",message)
                    # print("-- Consumer G --[%s]" % (self.name))
            time.sleep(1)

if __name__ == "__main__":
        # #-----------------------
    # Queue  提供了同步、线程安全的队列类
    # 队列原则: 先进先出  FIFO
    # 栈的原则: 先进后出  LIFO
    # 这些队列都实现了锁原语，原子操作
    #创建队列 ，这个队列只能用于线程，而进程中不能使用
    queue_name = Queue() # 队列
    # 队列初创
    for var in range(500):
        queue_name.put("》》初始产品"+ str(var))
    # 创建二次线程 producer
    for var in range(2):
        producer_name = Class_Producer_F()
        producer_name.start()
    # 创建二次线程 consumer
    for var in range(5):
        producer_name = Class_Consumer_G()
        producer_name.start()
    # 2 + 5 +1(主线程) = 8 （总线程）
result:

-----------------------
2.1 ThreadLocal

ThreadLocal不用查找dict，ThreadLocal帮你自动的做这个事：
全局变量Local_school就是一个ThreadLocal对象，
每个Thread对他都可以读写student属性，但互不影响。
你可以把ThreadLocal看出全局变量，但每个属性，
如，Local_school.student都是线程的局部变量，
可以任意读写而互不干扰，也不用管理锁的问题，ThreadLocal内部会处理。

可以理解为全局变量local_school是dict,不但可以用Local_school.student，
还可以绑定其他变量，如Local_school.teacher等。

ThreadLocal最常用的地方就是为每个线程绑定一个数据库连接，
HTTP请求，用户身份信息，这样一个线程的所有调用到的处理函数
都可以非常方便的访问这些资源。

总结：
ThreadLocal变量虽然是全局变量，但是每个线程都只能读写自己线程的独立副本，
互不干扰，ThreadLocal解决了参数在一个线程中各个函数之间互相传递的问题。


e.g:
# ThreadLocal
def Function_Process_student():
    # 获取当前线程关联的student
    student_name = local_school.student
    print("%s: (in %s)"% (student_name,threading.current_thread()))

def Function_Process_thread(name):
    local_var_name =name

    # 绑定ThreadLocal的student
    local_school.student = local_var_name
    Function_Process_student()

if __name__ == "__main__":
    # ThreadLocal
    # 创建全局ThreadLocal对象
    local_school = threading.local()

    thread_name_O = threading.Thread(target=Function_Process_thread, args=('图形分析',), name='Thread-A')
    thread_name_O.start()
    thread_name_O.join()

    thread_name_P = threading.Thread(target = Function_Process_thread, args=('数据处理',), name='Thread-B')
    thread_name_P.start()
    thread_name_P.join()

    thread_name_Q = threading.Thread(target = Function_Process_thread, args=('信号处理',), name='Thread-C')
    thread_name_Q.start()
    thread_name_Q.join()
result:

-----------------------
2.2 异步

e.g:
def Function_Test_G():
    print("进程池中的进程PID=[%s],PPID=[%s]" % (os.getpid(),os.getppid()))
    for var in range(3):
        print("-- [%s]" % (var))
        time.sleep(1)
    return "Function_Test_G"

def Function_Test_H(args):
    print("callback func PID =[%s]" % (os.getpid()))
    print("callback func args=[%s]" % (args))

if __name__ == "__main__":
    # 异步
    #创建进程池
    pool_name = Pool(3)
    # callback 回调函数 子线程结束后，立刻执行回调函数
    pool_name.apply_async(func = Function_Test_G,callback= Function_Test_H )

    # 异步的理解：主进程正在做某件事情，
    # 突然来了一件更需要立刻去做的事情，
    # 那么这种，在父进程去做某件事情的时候　
    # 并不知道是什么时候去做，的模式　就称为异步
    while True:
        time.sleep(1)
        print("主进程 PID = [%s]" % (os.getpid()))
result:
    进程池中的进程PID=[69792],PPID=[71624]
    -- [0]
    主进程 PID = [71624]
    -- [1]
    主进程 PID = [71624]
    -- [2]
    主进程 PID = [71624]
    callback func PID =[71624]
    callback func args=[Function_Test_G]
    主进程 PID = [71624]
    主进程 PID = [71624]
    主进程 PID = [71624]
    
-----------------------

----------------------------------------------

============================================================================
'''

#
'''
# ============================================================================
# Function:  
# Explain :  输入参数   
#         :  输出参数  
# ============================================================================
'''
# #-----------------------
# 单 / 多 线程执行

def Function_Say_Sorry():
    print("A say: Sorry !")
    time.sleep(1)

# #-----------------------
def Function_Sing():
    for var in range(10):
        print("正在唱歌 %d" % var)
        time.sleep(1)

def Function_Dance():
    for var in range(5):
        print("正在跳舞 %d" % var)
        time.sleep(1)

# #-----------------------
#线程执行代码的封装
class Class_MyThread(threading.Thread):
    def run(self):
        for var in range(10):
            time.sleep(1)
            message = "I am " + self.name + " @ " + str(var)
            print("message: %s" % (message))

# #-----------------------
# 线程的执行顺序
  #重复封装类
def Function_Test_A():
    for var in range(10):
        thread_name_local = Class_MyThread()
        thread_name_local.start()

# #-----------------------
# 多线程-共享全局变量

def Function_Work_A():
    global global_var_number_A

    for var in range(10):
        global_var_number_A += 1
    print("work A ,number is %d" % (global_var_number_A))


def Function_Work_B():
    global global_var_number_A

    print("work A ,number is %d" % (global_var_number_A))

# #-----------------------
# 列表当做实参传递到线程中
def Function_Work_C(nums):
    local_var_number = nums

    local_var_number.append("CC")
    print("work C ,number is %s" % (local_var_number))


def Function_Work_D(nums):
    local_var_number = nums
    time.sleep(1)
    print("work D ,number is %s" % (local_var_number))


# #-----------------------
# 同步
def Function_Test_B():
    global global_var_number
    for var in range(1000000):
        global_var_number += 1
    print("Test B ,number is %d" % (global_var_number))

def Function_Test_C():
    global global_var_number
    for var in range(1000000):
        global_var_number += 1
    print("Test C ,number is %d" % (global_var_number))

# #-----------------------
# 互斥锁
def Function_Test_D():
    global global_var_number
    # Test_D和Test_F线程都在抢着上锁，对这个锁，如果一方上锁成功，
    # 导致另一方会堵塞（一直等待）到这个锁被解锁为止
    mutexFlag = mutex.acquire(True)
    for var in range(1000000):
        # True 表示堵塞，如果这个锁在上锁之前已经被上锁了，那么这个线程会在
        #False 表示非堵塞，即不管本次调用能够成功上锁，都会卡在这，而继续
        if mutexFlag:
            global_var_number += 1
    # 用来对mutex指向的这个锁，进行解锁，只要解锁了，那么接下来会让所有
    # 因为这个锁，被上了锁而堵塞的线程进行抢着上锁
    mutex.release()

    print("Test D ,number is %d" % (global_var_number))

def Function_Test_E():
    global global_var_number

    for var in range(1000000):
        mutexFlag = mutex.acquire(True)
        if mutexFlag:
            global_var_number += 1
        mutex.release()

    print("Test E ,number is %d" % (global_var_number))

def Function_Test_F():
    global global_var_number
    mutexFlag = mutex.acquire(True)
    for var in range(1000000):
        if mutexFlag:
            global_var_number += 1
    mutex.release()

    print("Test F ,number is %d" % (global_var_number))

# #-----------------------
# 多线程-非共享数据
class Class_My_Thread(threading.Thread):
    # 重写 构造方法
    #   1. 全局变量在多个线程中　共享，为了保证正确运行需要锁
    #   2. 非全局变量在每个线程中　各有一份，不会共享，当然了不需要加锁
    def __init__(self,number,SleepTime):
        threading.Thread.__init__(self)
        self.num       = number
        self.SleepTime = SleepTime

    def run(self):
        self.num += 1  # 局部变量
        time.sleep(self.SleepTime)
        print("线程 (%s), number = (%s)" % (self.name,self.num))

# #-----------------------
# 死锁
class Class_My_Thread_A(threading.Thread):

    def run(self):
        if mutexA.acquire():
            print(self.name + "--do 1 up --")
            time.sleep(1)

            if mutexB.acquire(True,3):
                print(self.name + "--do 1 down --")
                mutexB.release()
            mutexA.release()

class Class_My_Thread_B(threading.Thread):

    def run(self):
        if mutexB.acquire(True,4):
            print(self.name + "--do 2 up --")
            time.sleep(1)

            if mutexA.acquire():
                print(self.name + "--do 2 down --")
                mutexA.release()
            mutexB.release()

# #-----------------------
# 多线程有序执行
class Class_Task_C(threading.Thread):
    def run(self):
        while True:
            if mutexC.acquire():
                print("--- Task C ---[%s]" % (self.name))
                time.sleep(1)
                mutexD.release()

class Class_Task_D(threading.Thread):
    def run(self):
        while True:
            if mutexD.acquire():
                print("--- Task D ---[%s]" % (self.name))
                time.sleep(1)
                mutexE.release()

class Class_Task_E(threading.Thread):
    def run(self):
        while True:
            if mutexE.acquire():
                print("--- Task E ---[%s]" % (self.name))
                time.sleep(1)
                mutexC.release()
# #-----------------------
# FIFO 队列  生产者消费模式
class Class_Producer_F(threading.Thread):
    def run(self):
        # global queue
        var_count = 0

        while True:
            if queue_name.qsize() < 1000 :
                # 每次就生产100个商品，二个线程做个事
                for var in range(100):
                    var_count += 1
                    message = "生成产品" + str(var_count)
                    queue_name.put(message)
                    # print("-- Producer F --[%s]" % (self.name))
            time.sleep(0.5)

class Class_Consumer_G(threading.Thread):
    def run(self):
        # global queue

        while True:
            if queue_name.qsize() > 100:
                # 若队列数量大于100，每个线程就取出3个商品，五个线程做这个事
                for var in range(3):
                    message = self.name + "消费了" + queue_name.get()
                    print("message: (%s)",message)
                    # print("-- Consumer G --[%s]" % (self.name))
            time.sleep(1)

# #-----------------------
# ThreadLocal
def Function_Process_student():
    # 获取当前线程关联的student
    student_name = local_school.student
    print("%s: (in %s)"% (student_name,threading.current_thread()))

def Function_Process_thread(name):
    local_var_name =name

    # 绑定ThreadLocal的student
    local_school.student = local_var_name
    Function_Process_student()

# #-----------------------
# 异步
def Function_Test_G():
    print("进程池中的进程PID=[%s],PPID=[%s]" % (os.getpid(),os.getppid()))
    for var in range(3):
        print("-- [%s]" % (var))
        time.sleep(1)
    return "Function_Test_G"

def Function_Test_H(args):
    print("callback func PID =[%s]" % (os.getpid()))
    print("callback func args=[%s]" % (args))

# #-----------------------

# #-----------------------

# ============================================================================
'''
# ============================================================================
#   测试专用
# ============================================================================
'''
if __name__ == "__main__":
    # #-----------------------
    # # 单线程执行
    # for var in range(10):
    #     Function_Say_Sorry()



    # # -----------------------
    # # 多线程执行
    # for var in range(10):
    #     # 创建线程
    #     thread_name = threading.Thread(target = Function_Say_Sorry)
    #     thread_name.start()



    # # # -----------------------
    # # 主线程等待子线程结束而结束
    # print("--- start ---:
 %s" % (time.time()))
    #
    # time_A = threading.Thread(target = Function_Sing)
    # time_B = threading.Thread(target = Function_Dance)
    # time_A.start()
    # time_B.start()
    # # time.sleep(5) #
    # print("--- finish ---:
%s" % (time.time()))
    #
    # # 查看线程数量
    # while True:
    #     length = len(threading.enumerate())
    #     print("当前运行的线程数为： [%d] " % (length))
    #     if length <= 1 :
    #         break
    #     time.sleep(0.5)



    # # # -----------------------
    # #线程执行代码的封装
    # thread_name = Class_MyThread()
    # thread_name.start()



    # # # -----------------------
    # #线程执行顺序
    # Function_Test_A()



    # # -----------------------
    # 多线程-共享全局变量
    # global_var_number_A = 100
    #
    # print("创建线程之前，number is %d" %(global_var_number_A))
    # thread_name_A = threading.Thread(target = Function_Work_A )
    # thread_name_A.start()
    # time.sleep(1)
    #
    # thread_name_B = threading.Thread(target = Function_Work_B )
    # thread_name_B.start()



    # # -----------------------
    # 列表当做实参传递到线程中
    # global_nums = [11,22,33,44,55]
    # thread_name_C = threading.Thread(target=Function_Work_C, args=(global_nums,))
    # thread_name_C.start()
    # thread_name_D = threading.Thread(target=Function_Work_D, args=(global_nums,))
    # thread_name_D.start()



    # # # -----------------------
    # # 同步
    # global_var_number = 0
    # thread_name_E = threading.Thread(target = Function_Test_B)
    # thread_name_E.start()
    # time.sleep(3)
    # thread_name_F = threading.Thread(target = Function_Test_C)
    # thread_name_F.start()
    #
    # print("number: %s" % (global_var_number))



    # # -----------------------
    # 互斥锁
    # global_var_number = 0
    # # 创建锁
    # mutex = threading.Lock()
    #
    # thread_name_D = threading.Thread(target=Function_Test_D)
    # thread_name_D.start()
    #
    # thread_name_E = threading.Thread(target = Function_Test_E)
    # thread_name_E.start()
    #
    # thread_name_F = threading.Thread(target = Function_Test_F)
    # thread_name_F.start()
    #
    # print("number: %s" % (global_var_number))


    # # # -----------------------
    # # 多线程-非共享数据
    # # 创建锁
    # mutex = threading.Lock()
    # thread_name_G = Class_My_Thread(100, 3)
    # thread_name_G.start()
    #
    # thread_name_H = Class_My_Thread(200, 1)
    # thread_name_H.start()
    #
    # thread_name_I = Class_My_Thread(300, 2)
    # thread_name_I.start()



    # # # -----------------------
    # # 死锁  不能执行的效果
    # mutexA = threading.Lock()
    # mutexB = threading.Lock()
    # thread_name_J = Class_My_Thread_A()
    # thread_name_J.start()
    # thread_name_K = Class_My_Thread_B()
    # thread_name_K.start()
    #
    # print("...")
    # #避免死锁
    # # 程序设计时，要尽量避免，银行家算法
    # # 添加超时时间等 acquire（timeout）很重要！！！


    # # #-----------------------
    # # 同步应用
    # # 创建锁
    # mutexC = threading.Lock()
    #
    # mutexD = threading.Lock()
    # mutexD.acquire()
    # mutexE = threading.Lock()
    # mutexE.acquire()
    #
    # thread_name_L = Class_Task_C()
    # thread_name_L.start()
    #
    # thread_name_M = Class_Task_D()
    # thread_name_M.start()
    #
    # thread_name_N = Class_Task_E()
    # thread_name_N.start()


    # # #-----------------------
    # # Queue  提供了同步、线程安全的队列类
    # # 队列原则: 先进先出  FIFO
    # # 栈的原则: 先进后出  LIFO
    # # 这些队列都实现了锁原语，原子操作
    # #创建队列 ，这个队列只能用于线程，而进程中不能使用
    # queue_name = Queue() # 队列
    # # 队列初创
    # for var in range(500):
    #     queue_name.put("》》初始产品"+ str(var))
    # # 创建二次线程 producer
    # for var in range(2):
    #     producer_name = Class_Producer_F()
    #     producer_name.start()
    # # 创建二次线程 consumer
    # for var in range(5):
    #     producer_name = Class_Consumer_G()
    #     producer_name.start()
    # # 2 + 5 +1(主线程) = 8 （总线程）



    # # #-----------------------
    # # ThreadLocal
    # # 创建全局ThreadLocal对象
    # local_school = threading.local()
    # thread_name_O = threading.Thread(target=Function_Process_thread, args=('图形分析',), name='Thread-A')
    # thread_name_O.start()
    # thread_name_O.join()
    # thread_name_P = threading.Thread(target = Function_Process_thread, args=('数据处理',), name='Thread-B')
    # thread_name_P.start()
    # thread_name_P.join()
    #
    # thread_name_Q = threading.Thread(target = Function_Process_thread, args=('信号处理',), name='Thread-C')
    # thread_name_Q.start()
    # thread_name_Q.join()


    # # #-----------------------
    # # 异步
    # #创建进程池
    # pool_name = Pool(3)
    # # callback 回调函数 子线程结束后，立刻执行回调函数
    # pool_name.apply_async(func = Function_Test_G,callback= Function_Test_H )
    #
    # # 异步的理解：主进程正在做某件事情，
    # # 突然来了一件更需要立刻去做的事情，
    # # 那么这种，在父进程去做某件事情的时候　
    # # 并不知道是什么时候去做，的模式　就称为异步
    # while True:
    #     time.sleep(1)
    #     print("主进程 PID = [%s]" % (os.getpid()))

    print("learn finish")
    

--

-----------------------------------------------------