Python_Example_Thread 线程 学习/经验/示例

   1 # coding=utf-8
   2 #---------------------------------
   3 ‘‘‘
   4 # Author  : chu ge 
   5 # Function: 线程 thread
   6 #
   7 ‘‘‘
   8 #---------------------------------
   9 ‘‘‘
  10 # --------------------------------
  11 # 导入模块 
  12 # 1.系统库
  13 # 2.第三方库
  14 # 3.相关定义库
  15 # --------------------------------
  16 ‘‘‘
  17 # 1.系统库
  18 import sys
  19 import os
  20 import time
  21 import random
  22 
  23 #2.第三方库
  24 
  25 #进程导入模块
  26 # from multiprocessing import Process
  27 from multiprocessing import Pool
  28 # from multiprocessing import Queue
  29 # from multiprocessing import Manager
  30 
  31 #进程导入模块
  32 import threading
  33 from threading import Thread
  34 from threading import Lock
  35 from queue import Queue
  36 
  37 
  38 
  39 
  40 
  41 
  42 #
  43 ‘‘‘
  44 ============================================================================
  45 #》》》》》》》》》》
  46 
  47 线程
  48 
  49 -----------------------
  50 python的thread模块比较底层，
  51 threading模块做了些包装，可以更加方便的被使用
  52 
  53 进程 VS 线程
  54 功能的不同
  55 进程，能够完成多任务，如一台电脑上可以同时运行多个微信
  56 线程，能够完成多任务，如一个微信界面中可以运行多个窗口
  57 定义的不同
  58 进程，是系统进行资源分配和调度的一个独立单位。
  59 线程，是进程的一个实体，是CPU调度和分派的基本单位，它是比进程更小的
  60 能独立运行的基本单位。
  61     线程自己基本不拥有系统资源，只拥有一点在运行中必不可少的资源，
  62     如：程序计数器，一组寄存器和栈，但是它可与同属一个进程的
  63     其他线程共享进程所拥有的全部的资源。
  64 
  65 区别
  66 一个程序至少有一个进程，一个进程至少有一个线程。
  67 线程的划分尺度小于进程，资源比进程少，使得多线程程序的并发性高。
  68 进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大提高程序的运行效率
  69 线程不能够独立执行，必须依存在进程中
  70 优缺点
  71 线程和进程在使用上各有优缺点：线程执行开销小但不利于资源管理和保护
  72 而进程正相反。
  73 
  74 ----------------------------------------------
  75 1. 线程 使用threading 模块
  76 
  77 -----------------------
  78 1.1 单 / 多线程执行
  79 e.g:
  80 #1如果多个线程执行同一函数，各是各的，互不影响
  81 def Function_Say_Sorry():
  82     print("A say: Sorry !")
  83     time.sleep(1)
  84 if __name__ == "__main__":
  85     # 单线程
  86     for var in range(10):
  87         Function_Say_Sorry()
  88 
  89     # 多线程执行
  90     for var in range(10):
  91         # 创建线程
  92         thread_name = threading.Thread(target = Function_Say_Sorry)
  93         thread_name.start()
  94 result:
  95 1)可以看出使用了多线程并发操作，花费时间要短很多
  96 2)创建好的线程，需要调用start()方法来启动
  97 
  98 -----------------------
  99 e.g:
 100 
 101 if __name__ == "__main__":
 102 
 103 result:
 104 
 105 -----------------------
 106 1.2 主线程会等待所有的子线程结束后才结束
 107 e.g:
 108 def Function_Sing():
 109     for var in range(10):
 110         print("正在唱歌 %d" % var)
 111         time.sleep(1)
 112 
 113 def Function_Dance():
 114     for var in range(5):
 115         print("正在跳舞 %d" % var)
 116         time.sleep(1)
 117         
 118 if __name__ == "__main__":
 119         # 主线程等待子线程结束而结束
 120     print("--- start ---:\n %s" % (time.time()))
 121 
 122     time_A = threading.Thread(target = Function_Sing)
 123     time_B = threading.Thread(target = Function_Dance)
 124     time_A.start()
 125     time_B.start()
 126     time.sleep(5) #
 127     print("--- finish ---:\n%s" % (time.time()))
 128     # 查看线程数量
 129     while True:
 130         length = len(threading.enumerate())
 131         print("当前运行的线程数为： [%d] " % (length))
 132         if length <= 1 :
 133             break
 134         time.sleep(0.5)
 135 result:
 136 
 137 -----------------------
 138 1.3 threading 
 139 1.3.1 线程执行代码的封装
 140 通过使用threading模块能完成多任务的程序开发，为了每个线程更完美
 141 所有使用threading模块时，往往定义新的子类class,
 142 只要继承threading.Thread就可以了，然后重写run方法
 143 
 144 e.g:
 145 #线程执行代码的封装
 146 class Class_MyThread(threading.Thread):
 147     def run(self):
 148         for var in range(10):
 149             time.sleep(1)
 150             message = "I am " + self.name + " @ " + str(var)
 151             print("message: %s" % (message))
 152 
 153 
 154 if __name__ == "__main__":
 155     thread_name = Class_MyThread()
 156     thread_name.start()
 157     
 158 result:
 159 python的threading.Thread类有一个run方法，
 160 用于定义线程的功能函数，可以在自己的线程类中覆盖还方法。
 161 而创建自己线程实例后，通过thread类start方法，可以启动该线程，
 162 交给python虚拟机进行调度，当该线程获得执行的机会时，就会调用run方法执行线程。
 163 
 164 -----------------------
 165 1.3.2 线程的执行顺序
 166 e.g:
 167 # 线程的执行顺序
 168   #重复封装类
 169 def Function_Test_A():
 170     for var in range(10):
 171         thread_name_local = Class_MyThread()
 172         thread_name_local.start()
 173 if __name__ == "__main__":
 174     Function_Test_A()
 175     
 176 result:
 177 从执行结果可以看出，多线程程序的执行顺序是不确定的
 178 当执行到sleep语句时，线程将会被阻塞(Blocked),到sleep结束后，
 179 线程进入就绪(Runnable)状态，等待调度，而线程调度就讲自行选择一个线程执行。
 180 上面代码可以保证运行完成的run,但线程启动顺序和run函数中每次循环的执行顺序都不确定。
 181 
 182 总结
 183 1） 每个线程一定会有一个name，尽管示例中没有指定线程对象的name，
 184 但是python会自动为线程指定一个名字。
 185 2)当线程的run函数结束时，该线程完成
 186 3)无法控制线程调度程序，但是可以通过别的方式来影响调度的方式
 187 4)线程的几种状态
 188 
 189      启动         调度        结束
 190 新建 -----> 就绪 <-----> 运行 -----> 死亡
 191             \            /
 192          满足条件       等待条件
 193                \      /
 194                等待(阻塞) 
 195 
 196 
 197 -----------------------
 198 1.4.1 多线程-共享全局变量
 199 
 200 e.g:
 201 def Function_Work_A():
 202     global global_var_number_A
 203 
 204     for var in range(10):
 205         global_var_number_A += 1
 206     print("work A ,number is %d" % (global_var_number_A))
 207 
 208 
 209 def Function_Work_B():
 210     global global_var_number_A
 211 
 212     print("work B ,number is %d" % (global_var_number_A))
 213 
 214 if __name__ == "__main__":
 215     global_var_number_A = 100
 216 
 217     print("创建线程之前，number is %d" %(global_var_number_A))
 218     thread_name_A = threading.Thread(target = Function_Work_A )
 219     thread_name_A.start()
 220     time.sleep(1)
 221 
 222     thread_name_B = threading.Thread(target = Function_Work_B )
 223     thread_name_B.start()
 224 
 225 result:
 226     创建线程之前，number is 100
 227     work A ,number is 110
 228     work B ,number is 110
 229 
 230 -----------------------
 231 1.4.2 列表当做实参传递到线程中
 232 e.g:
 233 # 列表当做实参传递到线程中
 234 def Function_Work_C(nums):
 235     local_var_number = nums
 236 
 237     local_var_number.append("CC")
 238     print("work C ,number is %s" % (local_var_number))
 239 
 240 
 241 def Function_Work_D(nums):
 242     local_var_number = nums
 243     time.sleep(1)
 244     print("work D ,number is %s" % (local_var_number))
 245 
 246 if __name__ == "__main__":
 247     global_nums = [11,22,33,44,55]
 248     thread_name_C = threading.Thread(target=Function_Work_C, args=(global_nums,))
 249     thread_name_C.start()
 250     thread_name_D = threading.Thread(target=Function_Work_D, args=(global_nums,))
 251     thread_name_D.start()
 252 
 253 result:
 254     work C ,number is [11, 22, 33, 44, 55, ‘CC‘]
 255     work D ,number is [11, 22, 33, 44, 55, ‘CC‘]
 256 总结：
 257 在一线程内的所有线程共享全局变量，能够在不适用其他方式的前提下，
 258 完成多线程之间的数据共享，这点要比多进程要好。
 259 缺点就是，线程是对全局变量随意遂改可能造成多线程之间，
 260 对全局变量的混乱，即线程非安全。
 261 
 262 -----------------------
 263 1.5 线程不安全  同步
 264 
 265 e.g:
 266 # 同步
 267 def Function_Test_B():
 268     global global_var_number
 269     for var in range(1000000):
 270         global_var_number += 1
 271     print("Test B ,number is %d" % (global_var_number))
 272 
 273 def Function_Test_C():
 274     global global_var_number
 275     for var in range(1000000):
 276         global_var_number += 1
 277     print("Test C ,number is %d" % (global_var_number))
 278 
 279 if __name__ == "__main__":
 280     global_var_number = 0
 281     thread_name_E = threading.Thread(target = Function_Test_B)
 282     thread_name_E.start()
 283     time.sleep(3)
 284     thread_name_F = threading.Thread(target = Function_Test_C)
 285     thread_name_F.start()
 286 
 287     print("number: %s" % (global_var_number))
 288 
 289 result:
 290     Test B ,number is 1000000
 291     number: 1059644
 292     Test C ,number is 2000000
 293 
 294 同步：就是协同步调，按照预定的先后次序进行运行。
 295 同指协同，协助，相互配合
 296 如，进程/线程同步，可以理解为进程/线程A和B一块配合，A执行到一定程度时
 297 要依靠B的某个结果，于是停下来，示意B运行，在将结果给A，A继续执行操作。
 298 
 299 解决问题：线程同步
 300 1.系统调用thread_name_E,然后获取到num的值为0，
 301 此时上一把锁，即不允许其他现在操作num
 302 2.对num的值进行+1
 303 3.解锁，此时num的值为1，其他的线程就可以使用num了，
 304 而且num的值是0而不是1
 305 4.同理，其他线程在对num进行修改时，都要先上锁，处理完成后再解锁。
 306 在上锁的整个过程中不允许其他线程访问，就保证了数据的正确性。
 307 
 308 -----------------------
 309 1.6 互斥锁
 310 
 311 当多个线程几乎同时修改某个共享数据的时候，需要进行同步控制。
 312 线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。
 313 
 314 互斥锁为资源引入一个状态： 锁定/非锁定
 315 
 316 某个线程要更改共享数据时，先将其锁定，此时资源的状态为锁定，
 317 其他线程不能更改，直到该线程释放资源，将资源的状态变成非锁定，
 318 其他的线程才能再次锁定该资源。
 319 互斥锁保证每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。
 320 
 321 
 322 threading模块定义Lock类，方便处理锁定：
 323 #创建锁
 324 mutex = threading.Lock()
 325 #锁定
 326 mutex.acquire([blocking])
 327 #释放
 328 mutex.release()
 329 
 330 在锁定方法中，acquire可以有个blocking参数
 331 如果设定blocking为True，则当前线程会阻塞，直到获得这个锁为止，如果没有指定，那么默认为True
 332 如果设定blocking为Fasle,则当前线程不会阻塞。
 333 
 334 
 335 上锁解锁过程：
 336 当一个线程调用锁的acpuire()方法获得解锁时，锁就进入Locked状态。
 337 每次只有一个线程可以获得锁。如果此时另外一个线程试图获得这个锁，
 338 该线程就会变为blocked状态，称为阻塞，直到拥有锁的线程调用锁release()方法释放后，锁进入unlocked状态
 339 线程调度程序从处于同步阻塞状态的线程中选择一个获得锁，并使得该线程进入运行(running)
 340 
 341 总结：
 342 锁的好处：
 343 确保了某段关键代码只能由一个线程从头到尾完整的执行
 344 锁的坏处：
 345 阻止了多线程并发执行，包含锁的某段代码实际上只能以单线程模式执行，效率降低了。
 346 由于可以存在多个锁，不同的线程持有不同的锁，并试图获得对方持有的锁，可能造成死锁。
 347 
 348 
 349 e.g:
 350 # 互斥锁
 351 def Function_Test_D():
 352     global global_var_number
 353     # Test_D和Test_F线程都在抢着上锁，对这个锁，如果一方上锁成功，
 354     # 导致另一方会堵塞（一直等待）到这个锁被解锁为止
 355     mutexFlag = mutex.acquire(True)
 356     for var in range(1000000):
 357         # True 表示堵塞，如果这个锁在上锁之前已经被上锁了，那么这个线程会在
 358         #False 表示非堵塞，即不管本次调用能够成功上锁，都会卡在这，而继续
 359         if mutexFlag:
 360             global_var_number += 1
 361     # 用来对mutex指向的这个锁，进行解锁，只要解锁了，那么接下来会让所有
 362     # 因为这个锁，被上了锁而堵塞的线程进行抢着上锁
 363     mutex.release()
 364 
 365     print("Test D ,number is %d" % (global_var_number))
 366 
 367 def Function_Test_E():
 368     global global_var_number
 369 
 370     for var in range(1000000):
 371         mutexFlag = mutex.acquire(True)
 372         if mutexFlag:
 373             global_var_number += 1
 374         mutex.release()
 375 
 376     print("Test E ,number is %d" % (global_var_number))
 377 
 378 def Function_Test_F():
 379     global global_var_number
 380     mutexFlag = mutex.acquire(True)
 381     for var in range(1000000):
 382         if mutexFlag:
 383             global_var_number += 1
 384     mutex.release()
 385 
 386     print("Test F ,number is %d" % (global_var_number))
 387 
 388 if __name__ == "__main__":
 389     global_var_number = 0
 390     # 创建锁
 391     mutex = threading.Lock()
 392 
 393     thread_name_D = threading.Thread(target=Function_Test_D)
 394     thread_name_D.start()
 395 
 396     thread_name_E = threading.Thread(target = Function_Test_E)
 397     thread_name_E.start()
 398 
 399     thread_name_F = threading.Thread(target = Function_Test_F)
 400     thread_name_F.start()
 401 
 402     print("number: %s" % (global_var_number))
 403     
 404 result:
 405 
 406 -----------------------
 407 1.7 多线程-非共享数据
 408 
 409 对于全局变量，在多线程中要格外小心，否则容易造成数据错乱的情况发生
 410 在多线程开发中，全局变量是多个线程都是共享的数据，而局部变量等是各自线程，是非共享的。
 411 
 412 e.g:
 413 class Class_My_Thread(threading.Thread):
 414     # 重写 构造方法
 415     def __init__(self,number,SleepTime):
 416         threading.Thread.__init__(self)
 417         self.num       = number
 418         self.SleepTime = SleepTime
 419 
 420     def run(self):
 421         self.num += 1  # 局部变量
 422         time.sleep(self.SleepTime)
 423         print("线程 (%s), number = (%s)" % (self.name,self.num))
 424 
 425 if __name__ == "__main__":
 426     # 创建锁
 427     mutex = threading.Lock()
 428     thread_name_G = Class_My_Thread(100, 3)
 429     thread_name_G.start()
 430 
 431     thread_name_H = Class_My_Thread(200, 1)
 432     thread_name_H.start()
 433 
 434     thread_name_I = Class_My_Thread(300, 2)
 435     thread_name_I.start()
 436     
 437 result:
 438     线程 (Thread-2), number = (201)
 439     线程 (Thread-3), number = (301)
 440     线程 (Thread-1), number = (101)
 441 局部变量是不共享数据的。
 442 
 443 -----------------------
 444 1.8  死锁
 445 
 446 在线程共享多个资源的时候，如果二个线程分别占有一部分资源
 447 并且同时等待对方的资源，就会造成死锁。
 448 尽管死锁很少发生，但一旦发生就会造成应用的停止响应。
 449 
 450 e.g:
 451 # 死锁
 452 class Class_My_Thread_A(threading.Thread):
 453     def run(self):
 454         if mutexA.acquire():
 455             print(self.name + "--do 1 up --")
 456             time.sleep(1)
 457 
 458             if mutexB.acquire():
 459                 print(self.name + "--do 1 down --")
 460                 mutexB.release()
 461             mutexA.release()
 462 
 463 
 464 class Class_My_Thread_B(threading.Thread):
 465     def run(self):
 466         if mutexB.acquire():
 467             print(self.name + "--do 2 up --")
 468             time.sleep(1)
 469 
 470             if mutexA.acquire():
 471                 print(self.name + "--do 2 down --")
 472                 mutexA.release()
 473             mutexB.release()
 474 
 475 if __name__ == "__main__":
 476     # 死锁  不能执行的效果
 477     mutexA = threading.Lock()
 478     mutexB = threading.Lock()
 479     thread_name_J = Class_My_Thread_A()
 480     thread_name_J.start()
 481     thread_name_K = Class_My_Thread_B()
 482     thread_name_K.start()
 483 
 484     print("...")
 485     
 486 result:
 487     #避免死锁
 488     # 程序设计时，要尽量避免，银行家算法
 489     # 添加超时时间等 acquire（timeout）很重要！！！
 490     
 491     
 492 -----------------------
 493 1.9.1 同步应用
 494 多线程有序执行
 495 
 496 可以使用互斥锁完成多个任务，有序的进程工作，这就是线程的同步
 497 
 498 e.g:
 499 # 多线程有序执行
 500 class Class_Task_C(threading.Thread):
 501     def run(self):
 502         while True:
 503             if mutexC.acquire():
 504                 print("--- Task C ---[%s]" % (self.name))
 505                 time.sleep(1)
 506                 mutexD.release()
 507 
 508 class Class_Task_D(threading.Thread):
 509     def run(self):
 510         while True:
 511             if mutexD.acquire():
 512                 print("--- Task D ---[%s]" % (self.name))
 513                 time.sleep(1)
 514                 mutexE.release()
 515 
 516 class Class_Task_E(threading.Thread):
 517     def run(self):
 518         while True:
 519             if mutexE.acquire():
 520                 print("--- Task F ---[%s]" % (self.name))
 521                 time.sleep(1)
 522                 mutexC.release()
 523                 
 524 if __name__ == "__main__":
 525     # 同步应用
 526     # 创建锁
 527     mutexC = threading.Lock()
 528 
 529     mutexD = threading.Lock()
 530     mutexD.acquire()
 531     mutexE = threading.Lock()
 532     mutexE.acquire()
 533 
 534     thread_name_L = Class_Task_C()
 535     thread_name_L.start()
 536 
 537     thread_name_M = Class_Task_D()
 538     thread_name_M.start()
 539 
 540     thread_name_N = Class_Task_E()
 541     thread_name_N.start()
 542 result:
 543     --- Task C ---[Thread-1]
 544     --- Task D ---[Thread-2]
 545     --- Task F ---[Thread-3]
 546     --- Task C ---[Thread-1]
 547     --- Task D ---[Thread-2]
 548     --- Task F ---[Thread-3]
 549     --- Task C ---[Thread-1]
 550     --- Task D ---[Thread-2]
 551 
 552 -----------------------
 553 2.0 Queue 队列
 554 
 555 Queue使用说明
 556 1) 对于Queue，在多线程通信之间扮演重要的角色
 557 2) 添加数据队列中，使用put()方法
 558 3) 从队列中获取数据，使用get()方法
 559 4) 判断队列中是否还有数据，使用qsize()方法
 560 
 561 生产者消费者模式说明：
 562 在线程世界里。生成者就是生产数据的线程，消费者就是消费数据的线程。
 563 在多线程开发中，如果生产者处理速度很快，而消费者处理速度很慢，
 564 那么生成者就必须等待消费者处理完，才能继续生产数据。同样道理
 565 如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。
 566 为了解决这个问题引入了生产者和消费者模式。
 567 
 568 生产者消费者模式，是通过一个容器来解决生产者和消费者的强耦合问题。
 569 生产者和消费者之间彼此之间不直接通讯，而通过阻塞队列来进行通讯，
 570 所有生产者生产数据之后不用等待消费者处理，直接扔给阻塞队列，
 571 消费者不找生产者要数据，而是直接从阻塞队列里取出数据，
 572 阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
 573 这个阻塞队列就是用来给生产者和消费者解耦的。
 574 纵观大多数设计模式，都会找到一个第三者出来进行解耦。
 575 
 576 
 577 e.g:
 578 class Class_Producer_F(threading.Thread):
 579     def run(self):
 580         # global queue
 581         var_count = 0
 582 
 583         while True:
 584             if queue_name.qsize() < 1000 :
 585                 # 每次就生产100个商品，二个线程做个事
 586                 for var in range(100):
 587                     var_count += 1
 588                     message = "生成产品" + str(var_count)
 589                     queue_name.put(message)
 590                     # print("-- Producer F --[%s]" % (self.name))
 591             time.sleep(0.5)
 592 
 593 class Class_Consumer_G(threading.Thread):
 594     def run(self):
 595         # global queue
 596 
 597         while True:
 598             if queue_name.qsize() > 100:
 599                 # 若队列数量大于100，每个线程就取出3个商品，五个线程做这个事
 600                 for var in range(3):
 601                     message = self.name + "消费了" + queue_name.get()
 602                     print("message: (%s)",message)
 603                     # print("-- Consumer G --[%s]" % (self.name))
 604             time.sleep(1)
 605 
 606 if __name__ == "__main__":
 607         # #-----------------------
 608     # Queue  提供了同步、线程安全的队列类
 609     # 队列原则: 先进先出  FIFO
 610     # 栈的原则: 先进后出  LIFO
 611     # 这些队列都实现了锁原语，原子操作
 612     #创建队列 ，这个队列只能用于线程，而进程中不能使用
 613     queue_name = Queue() # 队列
 614     # 队列初创
 615     for var in range(500):
 616         queue_name.put("》》初始产品"+ str(var))
 617     # 创建二次线程 producer
 618     for var in range(2):
 619         producer_name = Class_Producer_F()
 620         producer_name.start()
 621     # 创建二次线程 consumer
 622     for var in range(5):
 623         producer_name = Class_Consumer_G()
 624         producer_name.start()
 625     # 2 + 5 +1(主线程) = 8 （总线程）
 626 result:
 627 
 628 -----------------------
 629 2.1 ThreadLocal
 630 
 631 ThreadLocal不用查找dict，ThreadLocal帮你自动的做这个事：
 632 全局变量Local_school就是一个ThreadLocal对象，
 633 每个Thread对他都可以读写student属性，但互不影响。
 634 你可以把ThreadLocal看出全局变量，但每个属性，
 635 如，Local_school.student都是线程的局部变量，
 636 可以任意读写而互不干扰，也不用管理锁的问题，ThreadLocal内部会处理。
 637 
 638 可以理解为全局变量local_school是dict,不但可以用Local_school.student，
 639 还可以绑定其他变量，如Local_school.teacher等。
 640 
 641 ThreadLocal最常用的地方就是为每个线程绑定一个数据库连接，
 642 HTTP请求，用户身份信息，这样一个线程的所有调用到的处理函数
 643 都可以非常方便的访问这些资源。
 644 
 645 总结：
 646 ThreadLocal变量虽然是全局变量，但是每个线程都只能读写自己线程的独立副本，
 647 互不干扰，ThreadLocal解决了参数在一个线程中各个函数之间互相传递的问题。
 648 
 649 
 650 e.g:
 651 # ThreadLocal
 652 def Function_Process_student():
 653     # 获取当前线程关联的student
 654     student_name = local_school.student
 655     print("%s: (in %s)"% (student_name,threading.current_thread()))
 656 
 657 def Function_Process_thread(name):
 658     local_var_name =name
 659 
 660     # 绑定ThreadLocal的student
 661     local_school.student = local_var_name
 662     Function_Process_student()
 663 
 664 if __name__ == "__main__":
 665     # ThreadLocal
 666     # 创建全局ThreadLocal对象
 667     local_school = threading.local()
 668 
 669     thread_name_O = threading.Thread(target=Function_Process_thread, args=(‘图形分析‘,), name=‘Thread-A‘)
 670     thread_name_O.start()
 671     thread_name_O.join()
 672 
 673     thread_name_P = threading.Thread(target = Function_Process_thread, args=(‘数据处理‘,), name=‘Thread-B‘)
 674     thread_name_P.start()
 675     thread_name_P.join()
 676 
 677     thread_name_Q = threading.Thread(target = Function_Process_thread, args=(‘信号处理‘,), name=‘Thread-C‘)
 678     thread_name_Q.start()
 679     thread_name_Q.join()
 680 result:
 681 
 682 -----------------------
 683 2.2 异步
 684 
 685 e.g:
 686 def Function_Test_G():
 687     print("进程池中的进程PID=[%s],PPID=[%s]" % (os.getpid(),os.getppid()))
 688     for var in range(3):
 689         print("-- [%s]" % (var))
 690         time.sleep(1)
 691     return "Function_Test_G"
 692 
 693 def Function_Test_H(args):
 694     print("callback func PID =[%s]" % (os.getpid()))
 695     print("callback func args=[%s]" % (args))
 696 
 697 if __name__ == "__main__":
 698     # 异步
 699     #创建进程池
 700     pool_name = Pool(3)
 701     # callback 回调函数 子线程结束后，立刻执行回调函数
 702     pool_name.apply_async(func = Function_Test_G,callback= Function_Test_H )
 703 
 704     # 异步的理解：主进程正在做某件事情，
 705     # 突然来了一件更需要立刻去做的事情，
 706     # 那么这种，在父进程去做某件事情的时候　
 707     # 并不知道是什么时候去做，的模式　就称为异步
 708     while True:
 709         time.sleep(1)
 710         print("主进程 PID = [%s]" % (os.getpid()))
 711 result:
 712     进程池中的进程PID=[69792],PPID=[71624]
 713     -- [0]
 714     主进程 PID = [71624]
 715     -- [1]
 716     主进程 PID = [71624]
 717     -- [2]
 718     主进程 PID = [71624]
 719     callback func PID =[71624]
 720     callback func args=[Function_Test_G]
 721     主进程 PID = [71624]
 722     主进程 PID = [71624]
 723     主进程 PID = [71624]
 724     
 725 -----------------------
 726 
 727 ----------------------------------------------
 728 
 729 ============================================================================
 730 ‘‘‘
 731 
 732 #
 733 ‘‘‘
 734 # ============================================================================
 735 # Function:  
 736 # Explain :  输入参数   
 737 #         :  输出参数  
 738 # ============================================================================
 739 ‘‘‘
 740 # #-----------------------
 741 # 单 / 多 线程执行
 742 
 743 def Function_Say_Sorry():
 744     print("A say: Sorry !")
 745     time.sleep(1)
 746 
 747 # #-----------------------
 748 def Function_Sing():
 749     for var in range(10):
 750         print("正在唱歌 %d" % var)
 751         time.sleep(1)
 752 
 753 def Function_Dance():
 754     for var in range(5):
 755         print("正在跳舞 %d" % var)
 756         time.sleep(1)
 757 
 758 # #-----------------------
 759 #线程执行代码的封装
 760 class Class_MyThread(threading.Thread):
 761     def run(self):
 762         for var in range(10):
 763             time.sleep(1)
 764             message = "I am " + self.name + " @ " + str(var)
 765             print("message: %s" % (message))
 766 
 767 # #-----------------------
 768 # 线程的执行顺序
 769   #重复封装类
 770 def Function_Test_A():
 771     for var in range(10):
 772         thread_name_local = Class_MyThread()
 773         thread_name_local.start()
 774 
 775 # #-----------------------
 776 # 多线程-共享全局变量
 777 
 778 def Function_Work_A():
 779     global global_var_number_A
 780 
 781     for var in range(10):
 782         global_var_number_A += 1
 783     print("work A ,number is %d" % (global_var_number_A))
 784 
 785 
 786 def Function_Work_B():
 787     global global_var_number_A
 788 
 789     print("work A ,number is %d" % (global_var_number_A))
 790 
 791 # #-----------------------
 792 # 列表当做实参传递到线程中
 793 def Function_Work_C(nums):
 794     local_var_number = nums
 795 
 796     local_var_number.append("CC")
 797     print("work C ,number is %s" % (local_var_number))
 798 
 799 
 800 def Function_Work_D(nums):
 801     local_var_number = nums
 802     time.sleep(1)
 803     print("work D ,number is %s" % (local_var_number))
 804 
 805 
 806 # #-----------------------
 807 # 同步
 808 def Function_Test_B():
 809     global global_var_number
 810     for var in range(1000000):
 811         global_var_number += 1
 812     print("Test B ,number is %d" % (global_var_number))
 813 
 814 def Function_Test_C():
 815     global global_var_number
 816     for var in range(1000000):
 817         global_var_number += 1
 818     print("Test C ,number is %d" % (global_var_number))
 819 
 820 # #-----------------------
 821 # 互斥锁
 822 def Function_Test_D():
 823     global global_var_number
 824     # Test_D和Test_F线程都在抢着上锁，对这个锁，如果一方上锁成功，
 825     # 导致另一方会堵塞（一直等待）到这个锁被解锁为止
 826     mutexFlag = mutex.acquire(True)
 827     for var in range(1000000):
 828         # True 表示堵塞，如果这个锁在上锁之前已经被上锁了，那么这个线程会在
 829         #False 表示非堵塞，即不管本次调用能够成功上锁，都会卡在这，而继续
 830         if mutexFlag:
 831             global_var_number += 1
 832     # 用来对mutex指向的这个锁，进行解锁，只要解锁了，那么接下来会让所有
 833     # 因为这个锁，被上了锁而堵塞的线程进行抢着上锁
 834     mutex.release()
 835 
 836     print("Test D ,number is %d" % (global_var_number))
 837 
 838 def Function_Test_E():
 839     global global_var_number
 840 
 841     for var in range(1000000):
 842         mutexFlag = mutex.acquire(True)
 843         if mutexFlag:
 844             global_var_number += 1
 845         mutex.release()
 846 
 847     print("Test E ,number is %d" % (global_var_number))
 848 
 849 def Function_Test_F():
 850     global global_var_number
 851     mutexFlag = mutex.acquire(True)
 852     for var in range(1000000):
 853         if mutexFlag:
 854             global_var_number += 1
 855     mutex.release()
 856 
 857     print("Test F ,number is %d" % (global_var_number))
 858 
 859 # #-----------------------
 860 # 多线程-非共享数据
 861 class Class_My_Thread(threading.Thread):
 862     # 重写 构造方法
 863     #   1. 全局变量在多个线程中　共享，为了保证正确运行需要锁
 864     #   2. 非全局变量在每个线程中　各有一份，不会共享，当然了不需要加锁
 865     def __init__(self,number,SleepTime):
 866         threading.Thread.__init__(self)
 867         self.num       = number
 868         self.SleepTime = SleepTime
 869 
 870     def run(self):
 871         self.num += 1  # 局部变量
 872         time.sleep(self.SleepTime)
 873         print("线程 (%s), number = (%s)" % (self.name,self.num))
 874 
 875 # #-----------------------
 876 # 死锁
 877 class Class_My_Thread_A(threading.Thread):
 878 
 879     def run(self):
 880         if mutexA.acquire():
 881             print(self.name + "--do 1 up --")
 882             time.sleep(1)
 883 
 884             if mutexB.acquire(True,3):
 885                 print(self.name + "--do 1 down --")
 886                 mutexB.release()
 887             mutexA.release()
 888 
 889 class Class_My_Thread_B(threading.Thread):
 890 
 891     def run(self):
 892         if mutexB.acquire(True,4):
 893             print(self.name + "--do 2 up --")
 894             time.sleep(1)
 895 
 896             if mutexA.acquire():
 897                 print(self.name + "--do 2 down --")
 898                 mutexA.release()
 899             mutexB.release()
 900 
 901 # #-----------------------
 902 # 多线程有序执行
 903 class Class_Task_C(threading.Thread):
 904     def run(self):
 905         while True:
 906             if mutexC.acquire():
 907                 print("--- Task C ---[%s]" % (self.name))
 908                 time.sleep(1)
 909                 mutexD.release()
 910 
 911 class Class_Task_D(threading.Thread):
 912     def run(self):
 913         while True:
 914             if mutexD.acquire():
 915                 print("--- Task D ---[%s]" % (self.name))
 916                 time.sleep(1)
 917                 mutexE.release()
 918 
 919 class Class_Task_E(threading.Thread):
 920     def run(self):
 921         while True:
 922             if mutexE.acquire():
 923                 print("--- Task E ---[%s]" % (self.name))
 924                 time.sleep(1)
 925                 mutexC.release()
 926 # #-----------------------
 927 # FIFO 队列  生产者消费模式
 928 class Class_Producer_F(threading.Thread):
 929     def run(self):
 930         # global queue
 931         var_count = 0
 932 
 933         while True:
 934             if queue_name.qsize() < 1000 :
 935                 # 每次就生产100个商品，二个线程做个事
 936                 for var in range(100):
 937                     var_count += 1
 938                     message = "生成产品" + str(var_count)
 939                     queue_name.put(message)
 940                     # print("-- Producer F --[%s]" % (self.name))
 941             time.sleep(0.5)
 942 
 943 class Class_Consumer_G(threading.Thread):
 944     def run(self):
 945         # global queue
 946 
 947         while True:
 948             if queue_name.qsize() > 100:
 949                 # 若队列数量大于100，每个线程就取出3个商品，五个线程做这个事
 950                 for var in range(3):
 951                     message = self.name + "消费了" + queue_name.get()
 952                     print("message: (%s)",message)
 953                     # print("-- Consumer G --[%s]" % (self.name))
 954             time.sleep(1)
 955 
 956 # #-----------------------
 957 # ThreadLocal
 958 def Function_Process_student():
 959     # 获取当前线程关联的student
 960     student_name = local_school.student
 961     print("%s: (in %s)"% (student_name,threading.current_thread()))
 962 
 963 def Function_Process_thread(name):
 964     local_var_name =name
 965 
 966     # 绑定ThreadLocal的student
 967     local_school.student = local_var_name
 968     Function_Process_student()
 969 
 970 # #-----------------------
 971 # 异步
 972 def Function_Test_G():
 973     print("进程池中的进程PID=[%s],PPID=[%s]" % (os.getpid(),os.getppid()))
 974     for var in range(3):
 975         print("-- [%s]" % (var))
 976         time.sleep(1)
 977     return "Function_Test_G"
 978 
 979 def Function_Test_H(args):
 980     print("callback func PID =[%s]" % (os.getpid()))
 981     print("callback func args=[%s]" % (args))
 982 
 983 # #-----------------------
 984 
 985 # #-----------------------
 986 
 987 # ============================================================================
 988 ‘‘‘
 989 # ============================================================================
 990 #   测试专用
 991 # ============================================================================
 992 ‘‘‘
 993 if __name__ == "__main__":
 994     # #-----------------------
 995     # # 单线程执行
 996     # for var in range(10):
 997     #     Function_Say_Sorry()
 998 
 999 
1000 
1001     # # -----------------------
1002     # # 多线程执行
1003     # for var in range(10):
1004     #     # 创建线程
1005     #     thread_name = threading.Thread(target = Function_Say_Sorry)
1006     #     thread_name.start()
1007 
1008 
1009 
1010     # # # -----------------------
1011     # # 主线程等待子线程结束而结束
1012     # print("--- start ---:\n %s" % (time.time()))
1013     #
1014     # time_A = threading.Thread(target = Function_Sing)
1015     # time_B = threading.Thread(target = Function_Dance)
1016     # time_A.start()
1017     # time_B.start()
1018     # # time.sleep(5) #
1019     # print("--- finish ---:\n%s" % (time.time()))
1020     #
1021     # # 查看线程数量
1022     # while True:
1023     #     length = len(threading.enumerate())
1024     #     print("当前运行的线程数为： [%d] " % (length))
1025     #     if length <= 1 :
1026     #         break
1027     #     time.sleep(0.5)
1028 
1029 
1030 
1031     # # # -----------------------
1032     # #线程执行代码的封装
1033     # thread_name = Class_MyThread()
1034     # thread_name.start()
1035 
1036 
1037 
1038     # # # -----------------------
1039     # #线程执行顺序
1040     # Function_Test_A()
1041 
1042 
1043 
1044     # # -----------------------
1045     # 多线程-共享全局变量
1046     # global_var_number_A = 100
1047     #
1048     # print("创建线程之前，number is %d" %(global_var_number_A))
1049     # thread_name_A = threading.Thread(target = Function_Work_A )
1050     # thread_name_A.start()
1051     # time.sleep(1)
1052     #
1053     # thread_name_B = threading.Thread(target = Function_Work_B )
1054     # thread_name_B.start()
1055 
1056 
1057 
1058     # # -----------------------
1059     # 列表当做实参传递到线程中
1060     # global_nums = [11,22,33,44,55]
1061     # thread_name_C = threading.Thread(target=Function_Work_C, args=(global_nums,))
1062     # thread_name_C.start()
1063     # thread_name_D = threading.Thread(target=Function_Work_D, args=(global_nums,))
1064     # thread_name_D.start()
1065 
1066 
1067 
1068     # # # -----------------------
1069     # # 同步
1070     # global_var_number = 0
1071     # thread_name_E = threading.Thread(target = Function_Test_B)
1072     # thread_name_E.start()
1073     # time.sleep(3)
1074     # thread_name_F = threading.Thread(target = Function_Test_C)
1075     # thread_name_F.start()
1076     #
1077     # print("number: %s" % (global_var_number))
1078 
1079 
1080 
1081     # # -----------------------
1082     # 互斥锁
1083     # global_var_number = 0
1084     # # 创建锁
1085     # mutex = threading.Lock()
1086     #
1087     # thread_name_D = threading.Thread(target=Function_Test_D)
1088     # thread_name_D.start()
1089     #
1090     # thread_name_E = threading.Thread(target = Function_Test_E)
1091     # thread_name_E.start()
1092     #
1093     # thread_name_F = threading.Thread(target = Function_Test_F)
1094     # thread_name_F.start()
1095     #
1096     # print("number: %s" % (global_var_number))
1097 
1098 
1099     # # # -----------------------
1100     # # 多线程-非共享数据
1101     # # 创建锁
1102     # mutex = threading.Lock()
1103     # thread_name_G = Class_My_Thread(100, 3)
1104     # thread_name_G.start()
1105     #
1106     # thread_name_H = Class_My_Thread(200, 1)
1107     # thread_name_H.start()
1108     #
1109     # thread_name_I = Class_My_Thread(300, 2)
1110     # thread_name_I.start()
1111 
1112 
1113 
1114     # # # -----------------------
1115     # # 死锁  不能执行的效果
1116     # mutexA = threading.Lock()
1117     # mutexB = threading.Lock()
1118     # thread_name_J = Class_My_Thread_A()
1119     # thread_name_J.start()
1120     # thread_name_K = Class_My_Thread_B()
1121     # thread_name_K.start()
1122     #
1123     # print("...")
1124     # #避免死锁
1125     # # 程序设计时，要尽量避免，银行家算法
1126     # # 添加超时时间等 acquire（timeout）很重要！！！
1127 
1128 
1129     # # #-----------------------
1130     # # 同步应用
1131     # # 创建锁
1132     # mutexC = threading.Lock()
1133     #
1134     # mutexD = threading.Lock()
1135     # mutexD.acquire()
1136     # mutexE = threading.Lock()
1137     # mutexE.acquire()
1138     #
1139     # thread_name_L = Class_Task_C()
1140     # thread_name_L.start()
1141     #
1142     # thread_name_M = Class_Task_D()
1143     # thread_name_M.start()
1144     #
1145     # thread_name_N = Class_Task_E()
1146     # thread_name_N.start()
1147 
1148 
1149     # # #-----------------------
1150     # # Queue  提供了同步、线程安全的队列类
1151     # # 队列原则: 先进先出  FIFO
1152     # # 栈的原则: 先进后出  LIFO
1153     # # 这些队列都实现了锁原语，原子操作
1154     # #创建队列 ，这个队列只能用于线程，而进程中不能使用
1155     # queue_name = Queue() # 队列
1156     # # 队列初创
1157     # for var in range(500):
1158     #     queue_name.put("》》初始产品"+ str(var))
1159     # # 创建二次线程 producer
1160     # for var in range(2):
1161     #     producer_name = Class_Producer_F()
1162     #     producer_name.start()
1163     # # 创建二次线程 consumer
1164     # for var in range(5):
1165     #     producer_name = Class_Consumer_G()
1166     #     producer_name.start()
1167     # # 2 + 5 +1(主线程) = 8 （总线程）
1168 
1169 
1170 
1171     # # #-----------------------
1172     # # ThreadLocal
1173     # # 创建全局ThreadLocal对象
1174     # local_school = threading.local()
1175     # thread_name_O = threading.Thread(target=Function_Process_thread, args=(‘图形分析‘,), name=‘Thread-A‘)
1176     # thread_name_O.start()
1177     # thread_name_O.join()
1178     # thread_name_P = threading.Thread(target = Function_Process_thread, args=(‘数据处理‘,), name=‘Thread-B‘)
1179     # thread_name_P.start()
1180     # thread_name_P.join()
1181     #
1182     # thread_name_Q = threading.Thread(target = Function_Process_thread, args=(‘信号处理‘,), name=‘Thread-C‘)
1183     # thread_name_Q.start()
1184     # thread_name_Q.join()
1185 
1186 
1187     # # #-----------------------
1188     # # 异步
1189     # #创建进程池
1190     # pool_name = Pool(3)
1191     # # callback 回调函数 子线程结束后，立刻执行回调函数
1192     # pool_name.apply_async(func = Function_Test_G,callback= Function_Test_H )
1193     #
1194     # # 异步的理解：主进程正在做某件事情，
1195     # # 突然来了一件更需要立刻去做的事情，
1196     # # 那么这种，在父进程去做某件事情的时候　
1197     # # 并不知道是什么时候去做，的模式　就称为异步
1198     # while True:
1199     #     time.sleep(1)
1200     #     print("主进程 PID = [%s]" % (os.getpid()))
1201 
1202     print("learn finish")
1203