Author: 楚格
2018-11-17 17:34:58
IDE： Pycharm2018.02 Python 3.7
KeyWord : 线程 threading Thread
Explain:
-----------------------------------------------------
--
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 ---: %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 ---: %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 ---: %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 ---: %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
--
-----------------------------------------------------