Python 多任务1: 线程&多线程版UDP聊天器

一、线程介绍

• 1.1、线程，有时被称为轻量进程(Lightweight Process，LWP），是程序执行流的最小单元。一个标准的线程由线程ID，当前指令指针(PC），寄存器集合和堆栈组成。另外，线程是进程中的一个实体，是被系统独立调度和分派的基本单位，线程自己不拥有系统资源，只拥有一点儿在运行中必不可少的资源，但它可与同属一个进程的其它线程共享进程所拥有的全部资源。一个线程可以创建和撤消另一个线程，同一进程中的多个线程之间可以并发执行。由于线程之间的相互制约，致使线程在运行中呈现出间断性。线程也有就绪、阻塞和运行三种基本状态。就绪状态是指线程具备运行的所有条件，逻辑上可以运行，在等待处理机；运行状态是指线程占有处理机正在运行；阻塞状态是指线程在等待一个事件（如某个信号量），逻辑上不可执行。每一个程序都至少有一个线程，若程序只有一个线程，那就是程序本身。
• 1.2、线程是程序中一个单一的顺序控制流程。进程内有一个相对独立的、可调度的执行单元，是系统独立调度和分派CPU的基本单位指令运行时的程序的调度单位。在单个程序中同时运行多个线程完成不同的工作，称为多线程。

二、线程的使用

• 2.1、导入import threading,进行使用

  import threading
  import time
  
  
  def run():
  
       print("我要唱个歌")
       time.sleep(1)
  
  
  def main():
  
        for i in range(20):
            thread = threading.Thread(target=run)
            thread.start()
  
  
  if __name__ == '__main__':
  
         main()
  

• 2.2、有一个问题是线程在什么时候被创建，答案是：在调用.start() 之后，线程就会被创建并且运行线程

  import threading
  import time
  
  
  def test():
  
        for i in range(5):
  
            print("---%d---"%i)
            time.sleep(1)
  
  def main():
  
        print("在调用Thread之前打印当前线程的信息")
  
        print(threading.enumerate())
        thread1 = threading.Thread(target=test)
  
        print("在调用Thread之后打印当前线程的信息")
        print(threading.enumerate())
  
        thread1.start()
  
        print("在调用start之后打印当前线程的信息")
        print(threading.enumerate())
  
  
  if __name__ == '__main__':
  
        main()
  
  打印结果如下：
  在调用Thread之前打印当前线程的信息
  [<_MainThread(MainThread, started 140735879713664)>]
  在调用Thread之后打印当前线程的信息
  [<_MainThread(MainThread, started 140735879713664)>]
  ---0---
  在调用start之后打印当前线程的信息
  [<_MainThread(MainThread, started 140735879713664)>, <Thread(Thread-1, started 123145521430528)>]
  ---1---
  ---2---
  ---3---
  ---4---
  

• 2.3、通过上面使用threading模块能完成多任务的程序开发，为了让每个线程的封装性更完美，所以使用threading模块时，往往会定义一个新的子类class，只要继承threading.Thread就可以了，然后重写run方法

  import threading
  import time
  
  class MyThread(threading.Thread):
          def run(self):
                 for i in range(3):
                 time.sleep(1)
                 msg = "I'm "+self.name+' @ '+str(i) #name属性中保存的是当前线程的名字
                 print(msg)
  
  
  if __name__ == '__main__':
       t = MyThread()
       t.start()
  

  提示：python的threading.Thread类有一个run方法，用于定义线程的功能函数，可以在自己的线程类中覆盖该方法。而创建自己的线程实例后，通过Thread类的start方法，可以启动该线程，交给python虚拟机进行调度，当该线程获得执行的机会时，就会调用run方法执行线程。

• 2.4、线程的执行顺序

  import threading
  import time
  
  class MyThread(threading.Thread):
         def run(self):
              for i in range(3):
                    time.sleep(1)
                    msg = "I'm "+self.name+' @ '+str(i)
                    print(msg)
  def test():
         for i in range(5):
                t = MyThread()
                t.start()
  if __name__ == '__main__':
         test()
  

  执行结果：(运行的结果可能不一样，但是大体是一致的)

  I'm Thread-1 @ 0
  I'm Thread-2 @ 0
  I'm Thread-5 @ 0
  I'm Thread-3 @ 0
  I'm Thread-4 @ 0
  I'm Thread-3 @ 1
  I'm Thread-4 @ 1
  I'm Thread-5 @ 1
  I'm Thread-1 @ 1
  I'm Thread-2 @ 1
  I'm Thread-4 @ 2
  I'm Thread-5 @ 2
  I'm Thread-2 @ 2
  I'm Thread-1 @ 2
  I'm Thread-3 @ 2
  

  提示：从代码和执行结果我们可以看出，多线程程序的执行顺序是不确定的。当执行到sleep语句时，线程将被阻塞（Blocked），到sleep结束后，线程进入就绪（Runnable）状态，等待调度。而线程调度将自行选择一个线程执行。上面的代码中只能保证每个线程都运行完整个run函数，但是线程的启动顺序、run函数中每次循环的执行顺序都不能确定。

• 2.5、总结

  • 每个线程默认有一个名字，尽管上面的例子中没有指定线程对象的name，但是python会自动为线程指定一个名字。
  • 当线程的run()方法结束时该线程完成。
  • 无法控制线程调度程序，但可以通过别的方式来影响线程调度的方式。

三、多线程-共享全局变量

• 3.1、共享全局变量

  import threading
  import time
  
  # 定义一个全局变量
  num_value = 2
  
  def test1():
  
       global num_value
       num_value += 10
  
       print("----test1----num_value=%d"%num_value)
  
  
  def test2():
       print("----test2----num_value=%d" % num_value)
  
  
  def main():
  
       thread1 = threading.Thread(target=test1)
       thread2 = threading.Thread(target=test2)
  
       thread1.start()
       time.sleep(1)
  
       thread2.start()
       time.sleep(1)
  
       print("----main--thread---num_value=%d" % num_value)
  
  
  if __name__ == '__main__':
  
       main()
  

  打印结果是：

  ----test1----num_value=12
  ----test2----num_value=12
  ----main--thread---num_value=12
  

  提示：在一个函数中 对全局变量进行修改的时候，到底是否需要使用 global 进行说明，要看是否对全局变量的执行指向进行了修改；如果修改了执行，即让全局变量指向一个新的地方，那么必须使用 global,如果，仅仅是修改了 指向的空间中的数据，此时不用必须使用global;
  比如：对于不可变的全局变量：字符串、元组、常量等等，在函数内赋值的时候就是修改了其指向，那么就要加global，如果是可变的列表、字典在通过方法在函数内修改他们的值，就不需要加 global

• 3.2、多线程-共享全局变量-args参数

  from threading import Thread
  import time
  
  def work1(nums):
       nums.append(44)
       print("----in work1---",nums)
  
  
  def work2(nums):
       #延时一会，保证t1线程中的事情做完
       time.sleep(1)
       print("----in work2---",nums)
  
  g_nums = [11,22,33]
  
  t1 = Thread(target=work1, args=(g_nums,))
  t1.start()
  
  t2 = Thread(target=work2, args=(g_nums,))
  t2.start()
  

  打印结果：

  ----in work1--- [11, 22, 33, 44]
  ----in work2--- [11, 22, 33, 44]
  

  提示：t1 = Thread(target=work1, args=(g_nums,))线程调用可以传一个任意值：g_nums进去，args是一个元组

  • target: 指定将来 这个线程去哪个函数执行代码
  • args 指定将来调用 函数的时候 传递什么数据过去

• 3.3、创建线程是指定传递的参数、多线程共享全局变量的问题 (资源争夺的问题，如买票，银行取钱存钱的问题)，比如下面两个线程买 200万票，各买100万张票

  import threading
  import time
  
  num_ticket = 2000000
  
  
  def test1(num):
  
       global  num_ticket
  
       for i in range(num):
  
             num_ticket -= 1
       print("test1卖了%d张票" % num)
  
  
  def test2(num):
  
       global num_ticket
  
       for i in range(num):
             num_ticket -= 1
       print("test2卖了%d张票" % num)
  
  def main():
  
       thread1 = threading.Thread(target=test1,args=(1000000,))
       thread2 = threading.Thread(target=test2,args=(1000000,))
  
       thread1.start()
       thread2.start()
  
       time.sleep(5)
  
       # 等待上面两个线程执行完：也就是把票卖完，总共20万张票，各卖10万，最后应该还剩0张票
       print("还剩 %d 张票"%num_ticket)
  
  
  if __name__ == '__main__':
  
       main()
  

  打印结果是：

  test1卖了1000000张票
  test2卖了1000000张票
  还剩 550339 张票
  

  分析：(结果应该是0张)，造成这种结果的原因是，当test1和test2刚开始读到的总票数都是200万，test1卖一张是从 200万-1，而test2卖一张是从 200万-1，也就是读取的剩余的票数一样，都是从读取的票数减去自己卖去的票数，故造成这样的共享问题

四、多线程资源争夺的解决方案：线程同步技术(加锁)

• 4.1、同步的理解
  同步就是协同步调，按预定的先后次序进行运行，"同"字从字面上容易理解为一起动作，其实不是，"同"字应是指协同、协助、互相配合。如进程、线程同步，可理解为进程或线程A和B一块配合，A执行到一定程度时要依靠B的某个结果，于是停下来，示意B运行;B执行，再将结果给A;A再继续操作。

• 4.2、解决多线程同时修改全局变量的方式，也就是解决资源争夺的问题，思路，如下:

  • (1)、系统调用t1，然后获取到num_ticket的值为100万，此时上一把锁，即不允许其他线程操作num_ticket
  • (2)、test1对num_ticket的值进行-1
  • (3)、test1解锁，此时num_ticket的值为1999999，其他的线程就可以使用num_ticket了，而且是gnum_ticketnum的值不是200万而是1999999
  • (4)、同理其他线程在对num_ticket进行修改时，都要先上锁，处理完后再解锁，在上锁的整个过程中不允许其他线程访问，就保证了数据的正确性

• 4.3、互斥锁

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

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

  • threading模块中定义了Lock类，可以方便的处理锁定：

    # 创建锁
    mutex = threading.Lock()
    
    # 锁定(加锁🔐)
    mutex.acquire()
    
    # 释放(解锁🔓)
    mutex.release()
    

    提示：如果这个锁之前是没有上锁的，那么acquire不会堵塞
    如果在调用acquire对这个锁上锁之前 它已经被 其他线程上了锁，那么此时acquire会堵塞，直到这个锁被解锁为止

• 4.4、有了上面的思路，下面我们就是用 线程的互斥锁 来实现一下阻止多线程的资源争夺的问题，卖100万张票

  import threading
  import time
  
  # 定义一个全局的变量
  num_ticket = 2000000
  
  def test1(num):
  
       global  num_ticket
  
       for i in range(num):
             # 上锁，如果之前没有被上锁，那么此时上锁成功
             # 如果上锁之前已经被上锁，那么此时就会堵塞在这里，直到这个锁被解开位置
             thread_lock.acquire()
             num_ticket -= 1
             # 解锁
             thread_lock.release()
       print("test1卖了%d张票，还剩%d张票" % (num, num_ticket))
  
  
  def test2(num):
  
       global num_ticket
  
       for i in range(num):
             thread_lock.acquire()
             num_ticket -= 1
             thread_lock.release()
       print("test2卖了%d张票，还剩%d张票" %(num,num_ticket))
  
  # 创建一个互斥锁，默认是没有上锁的
  thread_lock = threading.Lock()
  
  
  def main():
  
       thread1 = threading.Thread(target=test1,args=(1000000,))
       thread2 = threading.Thread(target=test2,args=(1000000,))
  
       thread1.start()
       thread2.start()
  
       time.sleep(5)
  
       # 等待上面两个线程执行完：也就是把票卖完，总共20万张票，各卖10万，最后应该还剩0张票
       print("还剩 %d 张票"%num_ticket)
  
  
  if __name__ == '__main__':
  
        main()
  

  打印结果是：

  test2卖了1000000张票，还剩14799张票
  test1卖了1000000张票，还剩0张票
  还剩 0 张票
  

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

• 4.5、总结

  • 锁的好处：确保了某段关键代码只能由一个线程从头到尾完整地执行
  • 锁的坏处：(1)、阻止了多线程并发执行，包含锁的某段代码实际上只能以单线程模式执行，效率就大大地下降了；(2)、由于可以存在多个锁，不同的线程持有不同的锁，并试图获取对方持有的锁时，可能会造成死锁。

五、死锁

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

• 5.2、看下面造成死锁的例子

  import threading
  import time
  
  class MyThread1(threading.Thread):
           def run(self):
                # 对mutexA上锁
                mutexA.acquire()
  
                # mutexA上锁后，延时1秒，等待另外那个线程 把mutexB上锁
                print(self.name+'----do1---up----')
                time.sleep(1)
  
                # 此时会堵塞，因为这个mutexB已经被另外的线程抢先上锁了
                mutexB.acquire()
                print(self.name+'----do1---down----')
                mutexB.release()
  
                # 对mutexA解锁
                mutexA.release()
  
  class MyThread2(threading.Thread):
           def run(self):
                # 对mutexB上锁
                mutexB.acquire()
  
                # mutexB上锁后，延时1秒，等待另外那个线程 把mutexA上锁
                print(self.name+'----do2---up----')
                time.sleep(1)
  
                # 此时会堵塞，因为这个mutexA已经被另外的线程抢先上锁了
                mutexA.acquire()
                print(self.name+'----do2---down----')
                mutexA.release()
  
                # 对mutexB解锁
                mutexB.release()
  
  mutexA = threading.Lock()
  mutexB = threading.Lock()
  
  if __name__ == '__main__':
             t1 = MyThread1()
             t2 = MyThread2()
             t1.start()
             t2.start()
  

  运行结果：此时已经进入到了死锁状态，可以使用ctrl-c退出
  

  死锁的运行结果

• 5.3、 避免死锁

  • 程序设计时要尽量避免（银行家算法）
  • 添加超时时间等

六、多线程版UDP聊天器

多线程版UDP聊天器

• 6.1、流程

  • 创建套接字
  • 绑定本地信息
  • 输入对方的ip与端口
  • 创建子线程进行发送和接收数据

• 6.2、具体的代码

  import socket
  import threading
  
  
  def receive_message(udp_socket):
           """接收消息"""
  
        while True:
              receive_data = udp_socket.recvfrom(1024)
              print("接收的内容是：%s"%receive_data[0].decode("utf-8"))
  
  
  def send_message(udp_socket,other_ip,other_port):
           """发送消息"""
  
        while True:
  
              send_data = input("请输入发送的内容：")
              udp_socket.sendto(send_data.encode("utf-8"),(other_ip,other_port))
              print("你发的内容是：%s" % send_data)
  
  
  def main():
  
        # 1、创建
        udp_socket = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
  
        # 2、绑定本地端口信息
        local_addr = ("192.168.3.230",7990)
        udp_socket.bind(local_addr)
  
        # 3、输入对方的IP与端口号
        other_ip = input("请输入对方的ip：")
        other_port = int(input("亲输入对方的端口号port："))
  
        # 4、创建2个子线程用来 发送消息 和 接收消息
        thread_send = threading.Thread(target=send_message,args=(udp_socket,other_ip,other_port))
        thread_receive = threading.Thread(target=receive_message,args=(udp_socket,))
        thread_send.start()
        thread_receive.start()
  
  
  if __name__ == '__main__':
  
        main()