python【系列教程】之并发编程

一、线程的创建和启动

  

   

    

     

    
    

     

      import threading
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      # 编写一个普通的函数，作为线程执行体
     
    
   

    

     

    
    

     

      def action(max):
     
    
   

    

     

    
    

     
 for i in range(max):
     
    
   

    

     

    
    

     

       print(threading.current_thread().getName() + " " + str(i))
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      for i in range(100):
     
    
   

    

     

    
    

     

       print(threading.current_thread().getName() + " " + str(i))
     
    
   

    

     

    
    

     
 if i == 20:
     
    
   

    

     

    
    

     
 # 创建线程1
     
    
   

    

     

    
    

     

       t1 = threading.Thread(target=action, args=(100,))
     
    
   

    

     

    
    

     

       t1.start()
     
    
   

    

     

    
    

     
 # 创建线程2
     
    
   

    

     

    
    

     

       t2 = threading.Thread(target=action, args=(100,))
     
    
   

    

     

    
    

     

       t2.start()
     
    
   

    

     

    
    

     

      print('主线程执行完成')
     
    
   

    

     

    
    

     
 
     
    
  
 

  

   

    

     

    
    

     

      # 继承thread类创建线程类
     
    
   

    

     

    
    

     

      import threading
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      class FKThread(threading.Thread):
     
    
   

    

     

    
    

     
 def __init__(self):
     
    
   

    

     

    
    

     

       threading.Thread.__init__(self)
     
    
   

    

     

    
    

     
 self.i = 0
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 # 重写run方法
     
    
   

    

     

    
    

     
 def run(self):
     
    
   

    

     

    
    

     
 while self.i < 100:
     
    
   

    

     

    
    

     

       print(threading.current_thread().getName() + " " + str(self.i))
     
    
   

    

     

    
    

     
 self.i += 1
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      # 下面是主线程
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      for i in range(100):
     
    
   

    

     

    
    

     

       print(threading.current_thread().getName() + " " + str(i))
     
    
   

    

     

    
    

     
 if i == 20:
     
    
   

    

     

    
    

     

       ft1 = FKThread()
     
    
   

    

     

    
    

     

       ft1.start()
     
    
   

    

     

    
    

     

       ft2 = FKThread()
     
    
   

    

     

    
    

     

       ft2.start()
     
    
   

    

     

    
    

     

      print('主线程执行完成')
     
    
  
 

二、线程的生命周期

三、控制线程

  

   

    

     

    
    

     

      # join线程
     
    
   

    

     

    
    

     

      import threading
     
    
   

    

     

    
    

     

      def action(max):
     
    
   

    

     

    
    

     
 for i in range(max):
     
    
   

    

     

    
    

     

       print(threading.current_thread().name + " " + str(i))
     
    
   

    

     

    
    

     

      threading.Thread(target=action, args=(100,), name='新线程').start()
     
    
   

    

     

    
    

     

      for i in range(100):
     
    
   

    

     

    
    

     
 if i == 20:
     
    
   

    

     

    
    

     

       jt = threading.Thread(target=action, args=(100,), name="被join的线程")
     
    
   

    

     

    
    

     

       jt.start()
     
    
   

    

     

    
    

     

       jt.join()
     
    
   

    

     

    
    

     

       print(threading.current_thread().name + " " + str(i))
     
    
  
 

  

   

    

     

    
    

     

      #后台守护线程
     
    
   

    

     

    
    

     

      import threading
     
    
   

    

     

    
    

     

      def action(max):
     
    
   

    

     

    
    

     
 for i in range(max):
     
    
   

    

     

    
    

     

       print(threading.current_thread().name + " " + str(i))
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      t = threading.Thread(target=action,args=(100,),name='守护线程')
     
    
   

    

     

    
    

     

      #将线程设置为守护线程
     
    
   

    

     

    
    

     

      t.daemon = True
     
    
   

    

     

    
    

     

      t.start()
     
    
  
 

  

   

    

     

    
    

     

      #线程睡眠 sleep
     
    
   

    

     

    
    

     

      import  time
     
    
   

    

     

    
    

     

      for i in range(10):
     
    
   

    

     

    
    

     
 print('当前时间:%s' % time.ctime())
     
    
   

    

     

    
    

     

       #休眠1秒
     
    
   

    

     

    
    

     
 time.sleep(1)
     
    
  
 

四、线程同步

  

   

    

     

    
    

     

      class Account:
     
    
   

    

     

    
    

     
 # 构造方法
     
    
   

    

     

    
    

     
 def __init__(self, account_no, balance):
     
    
   

    

     

    
    

     
 self.account_no = account_no
     
    
   

    

     

    
    

     
 self._balance = balance
     
    
   

    

     

    
    

     
 # 获取一个可重入锁
     
    
   

    

     

    
    

     
 self.lock = threading.RLock
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 def getBalance(self):
     
    
   

    

     

    
    

     
 return self.balance
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 def draw(self, draw_aount):
     
    
   

    

     

    
    

     
 # 上锁
     
    
   

    

     

    
    

     
 self.lock.acquire()
     
    
   

    

     

    
    

     
 try:
     
    
   

    

     

    
    

     
 if self._balance >= draw_aount:
     
    
   

    

     

    
    

     

       print(threading.current_thread().name + "取钱成功！吐出钞票" + str(draw_aount))
     
    
   

    

     

    
    

     

       time.sleep(0.001)
     
    
   

    

     

    
    

     
 self._balance -= draw_aount
     
    
   

    

     

    
    

     

       print('余额：' + str(self._balance))
     
    
   

    

     

    
    

     
 else:
     
    
   

    

     

    
    

     

       print(threading.current_thread().name + "取钱失败，余额不足！" + str(draw_aount))
     
    
   

    

     

    
    

     

       except Exception as e:
     
    
   

    

     

    
    

     

       print(e)
     
    
   

    

     

    
    

     
 finally:
     
    
   

    

     

    
    

     
 # 释放锁
     
    
   

    

     

    
    

     
 self.lock.release()
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      def draw(account,draw_aount):
     
    
   

    

     

    
    

     

       account.draw(draw_aount)
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      acct = Account('1234567',1000)
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      threading.Thread(name='甲',target=draw,args=(acct,800)).start()
     
    
   

    

     

    
    

     

      threading.Thread(name='乙',target=draw,args=(acct,800)).start()
     
    
  
 

五、线程通信

  

   

    

     

    
    

     

      class Account:
     
    
   

    

     

    
    

     
 # 构造方法
     
    
   

    

     

    
    

     
 def __init__(self, account_no, balance):
     
    
   

    

     

    
    

     
 self.account_no = account_no
     
    
   

    

     

    
    

     
 self._balance = balance
     
    
   

    

     

    
    

     
 self.cond = threading.Condition()
     
    
   

    

     

    
    

     
 # 定义是否已经存钱的标识
     
    
   

    

     

    
    

     
 self._flag = False
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 def getBalance(self):
     
    
   

    

     

    
    

     
 return self.balance
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 def draw(self, draw_aount):
     
    
   

    

     

    
    

     
 # 上锁
     
    
   

    

     

    
    

     
 self.cond.acquire()
     
    
   

    

     

    
    

     
 try:
     
    
   

    

     

    
    

     
 # 如果flag是False，表明账户中还没有存钱，取钱方法被阻塞
     
    
   

    

     

    
    

     
 if not self._flag:
     
    
   

    

     

    
    

     
 self.cond.wait()
     
    
   

    

     

    
    

     
 else:
     
    
   

    

     

    
    

     

       print(threading.current_thread().name + "取钱：" + str(draw_aount))
     
    
   

    

     

    
    

     
 self._balance -= draw_aount
     
    
   

    

     

    
    

     

       print("账户余额为：" + str(self._balance))
     
    
   

    

     

    
    

     
 self._flag = False
     
    
   

    

     

    
    

     
 self.cond.notify_all()
     
    
   

    

     

    
    

     

       except Exception as e:
     
    
   

    

     

    
    

     

       print(e)
     
    
   

    

     

    
    

     
 finally:
     
    
   

    

     

    
    

     
 # 释放锁
     
    
   

    

     

    
    

     
 self.cond.release()
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 def deposit(self, deposit_amount):
     
    
   

    

     

    
    

     
 self.cond.acquire()
     
    
   

    

     

    
    

     
 try:
     
    
   

    

     

    
    

     
 # 如果flag标识是True，则说明账户有钱，存钱方法被阻塞
     
    
   

    

     

    
    

     
 if self._flag:
     
    
   

    

     

    
    

     
 self.cond.wait()
     
    
   

    

     

    
    

     
 else:
     
    
   

    

     

    
    

     

       print(threading.current_thread().name + "存款" + str(deposit_amount))
     
    
   

    

     

    
    

     
 self._balance += deposit_amount
     
    
   

    

     

    
    

     
 self._flag = True
     
    
   

    

     

    
    

     
 self.cond.notify_all()
     
    
   

    

     

    
    

     

       except Exception as e:
     
    
   

    

     

    
    

     

       print(e)
     
    
   

    

     

    
    

     
 finally:
     
    
   

    

     

    
    

     
 self.cond.release()
     
    
  
 

  

   

    

     

    
    

     

      #使用queue队列控制线程通信
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      import  threading
     
    
   

    

     

    
    

     

      import  time
     
    
   

    

     

    
    

     

      import queue
     
    
   

    

     

    
    

     

      def product(bq):
     
    
   

    

     

    
    

     

       str_tupln = ('python','kotlin','swift')
     
    
   

    

     

    
    

     
 for i in range(99999):
     
    
   

    

     

    
    

     

       print(threading.current_thread().name+"生产者准备生产元组元素")
     
    
   

    

     

    
    

     

       time.sleep(0.2)
     
    
   

    

     

    
    

     

       bq.put(str_tupln[i%3])
     
    
   

    

     

    
    

     

       print(threading.current_thread().name+"生产者生产元素完成！")
     
    
   

    

     

    
    

     

      def consume(bq):
     
    
   

    

     

    
    

     
 while True:
     
    
   

    

     

    
    

     

       print(threading.current_thread().name+"消费者准备消费元组元素")
     
    
   

    

     

    
    

     

       time.sleep(0.2)
     
    
   

    

     

    
    

     

       t = bq.get()
     
    
   

    

     

    
    

     

       print(threading.current_thread().name+"消费者消费【%s】元素完成！" % t)
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      bq = queue.Queue(maxsize=1)
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      #启动三个线程生产元素
     
    
   

    

     

    
    

     

      threading.Thread(name='1',target=product,args=(bq,)).start()
     
    
   

    

     

    
    

     

      threading.Thread(name='2',target=product,args=(bq,)).start()
     
    
   

    

     

    
    

     

      threading.Thread(name='3',target=product,args=(bq,)).start()
     
    
   

    

     

    
    

     

      #启动一个线程消费元素
     
    
   

    

     

    
    

     

      threading.Thread(name='4',target=consume,args=(bq,)).start()
     
    
  
 

六、线程池

  

   

    

     

    
    

     

      from concurrent.futures import ThreadPoolExecutor
     
    
   

    

     

    
    

     

      import threading
     
    
   

    

     

    
    

     

      import time
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      def action(max):
     
    
   

    

     

    
    

     

       my_sum = 0
     
    
   

    

     

    
    

     
 for i in range(max):
     
    
   

    

     

    
    

     

       print(threading.current_thread().name + " " + str(i))
     
    
   

    

     

    
    

     

       my_sum += 1
     
    
   

    

     

    
    

     
 return my_sum
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      # 创建一个包含两个线程的线程池
     
    
   

    

     

    
    

     

      pool = ThreadPoolExecutor(max_workers=2)
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      f1 = pool.submit(action, 50)
     
    
   

    

     

    
    

     

      f2 = pool.submit(action, 50)
     
    
   

    

     

    
    

     

      # 判断f1代表的任务是否结束
     
    
   

    

     

    
    

     

      print(f1.done)
     
    
   

    

     

    
    

     

      time.sleep(3)
     
    
   

    

     

    
    

     

      # 判断f2代表的任务是否结束
     
    
   

    

     

    
    

     

      print(f2.done)
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      #调用result方法会阻塞当前任务，直至当前任务完成
     
    
   

    

     

    
    

     

      print(f1.result(), f2.result())
     
    
   

    

     

    
    

     

      #关闭线程池
     
    
   

    

     

    
    

     

      pool.shutdown()
     
    
  
 

七、线程相关类

  

   

    

     

    
    

     

      #定时器
     
    
   

    

     

    
    

     

      from threading import  Timer
     
    
   

    

     

    
    

     

      def hello():
     
    
   

    

     

    
    

     

       print("hello world")
     
    
   

    

     

    
    

     

      t = Timer(10.0,hello)
     
    
   

    

     

    
    

     

      t.start()
     
    
  
 

  

   

    

     

    
    

     

      from threading import  Timer
     
    
   

    

     

    
    

     

      import time
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      count  = 0
     
    
   

    

     

    
    

     

      def print_time():
     
    
   

    

     

    
    

     
 print("当前时间:"+time.ctime())
     
    
   

    

     

    
    

     

       global  t,count
     
    
   

    

     

    
    

     
 count+=1
     
    
   

    

     

    
    

     
 if count < 10:
     
    
   

    

     

    
    

     

       t = Timer(1,print_time)
     
    
   

    

     

    
    

     

       t.start()
     
    
   

    

     

    
    

     

      t = Timer(1,print_time)
     
    
   

    

     

    
    

     

      t.start()
     
    
  
 

  

   

    

     

    
    

     

      import sched,time
     
    
   

    

     

    
    

     

      import  threading
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      #定义线程调度器
     
    
   

    

     

    
    

     

      s = sched.scheduler()
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      def print_time(name='default'):
     
    
   

    

     

    
    

     

       print('%s的时间：%s' % (name,time.ctime()))
     
    
   

    

     

    
    

     

      print('主线程',time.ctime())
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      #指定10秒后执行print_time函数
     
    
   

    

     

    
    

     

      s.enter(10,1,print_time())
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      s.enter(5,2,print_time,argument=('位置参数',))
     
    
   

    

     

    
    

     

      s.enter(5,1,print_time,kwargs={'name':'关键字参数'})
     
    
   

    

     

    
    

     

      s.run()
     
    
   

    

     

    
    

     

      print('主线程',time.ctime())
     
    
  
 

八、多进程

文章来源: blog.csdn.net，作者：血煞风雨城2018，版权归原作者所有，如需转载，请联系作者。

原文链接：blog.csdn.net/qq_31905135/article/details/101020299