初学Python——线程-低调大师

初学Python——线程

2018-07-27 523

什么是线程？

线程是进程内的独立的运行线路，是操作系统能够进行运算调度的最小单位，同时也是处理器调度的最小单位。线程被包含在进程之内，是进程中实际运作单位。

一个线程指的是进程中的一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。

Python中如何使用多线程？

首先要import threading

线程有两种调用方式，第一种是直接调用

import threading
import time
# 直接调用

def run(n):
    print("task {0} {1}".format(n,threading.current_thread()))
    time.sleep(2)

if __name__ == "__main__":

    t1 = threading.Thread(target=run,args=("t1",)) # 参数括号内的逗号不能省略
    t2 = threading.Thread(target=run,args=("t2",)) # 生成一个线程对象

    t1.start() # 启动线程
    t2.start()

        # 现在同时启动50个线程
    start = time.time()
    for i in range(50):
        t= threading.Thread(target=run,args=("t{0}".format(i),))
        t.start()
    end = time.time()
    cost = end - start
    print("cost time:",cost)

View Code

第二种是继承式

import threading
import time
'''继承式调用'''

class MyThread(threading.Thread):
    def __init__(self,n):
        super(MyThread,self).__init__()
        self.n = n

    def run(self):
        print("running task{0}  {1}".format(self.n,threading.current_thread()))
        time.sleep(1)


if __name__ == "__main__":

    start = time.time()
    obj = []

    for i in range(50):
        t= MyThread("t{0}".format(i))
        t.start()

    end = time.time()
    cost = end - start
    print("cost time:",cost)

继承式

使用多线程对比函数调用，得出结论是：线程是并发执行的（同时执行），而函数调用只能是顺序执行，多线程执行大大提高了运行效率。

但是以上两个程序在运行的时候发现了一个共同的问题：拿继承式的代码来说，理论上50个线程执行完需要经过一秒多才会执行完毕，退出程序，但是实际情况却是开启50个线程之后立马就退出程序，执行时间不足0.01s。为什么呢？

　　答案是多线程。当前运行的线程是主线程，主线程启动了50个子线程，启动完毕后继续做自己的事情，每个线程之间互不干扰，并行运行。因此无法用此方式测定50个程序到底运行了多久。

　　主线程运行完毕即退出程序，如果不特殊处理，它不会等待子线程处理完毕。所以我们如果想要等待子线程的运行结果，需要加上join()语句。

import threading
import time
'''继承式调用'''

class MyThread(threading.Thread):
    def __init__(self,n):
        super(MyThread,self).__init__()
        self.n = n

    def run(self):
        print("running task{0}  {1}".format(self.n,threading.current_thread()))
        time.sleep(1)


if __name__ == "__main__":

    start = time.time()
    obj = []

    for i in range(50):
        t= MyThread("t{0}".format(i))
        t.start()
        obj.append(t) # 为了不阻塞后面线程的启动，现将其加入列表

    for j in obj:
        j.join()
    print(threading.current_thread())  # 证明主线程
    end = time.time()
    cost = end - start
    print("cost time:",cost)

join等待

当线程被启动后，如果调用join()方法，则会在此线程执行完之前程序不往下走，也就是阻塞当前程序，使程序变为串行执行，这当然不是我们愿意看到的。在此程序中，我们希望它并发执行的同时，满足主线程等待所有子线程执行完毕后再结束这个条件。只需要在开启所有线程之后，一一地join()即可。

线程锁

在线程锁之间讲一下GIL全局性解释器锁。

首先要明白GIL并不是Python的特性，它是在实现Python解释器（CPython）时引入的一个概念。简单来说，GIL锁的存在使得：无论你开启多少个线程，无论你的CPU有多少核，在执行程序的时候同一时间只会占用一个核。所以，你以为的同时占用多核只是假象。再说一遍，这不是Python的特性，这只是CPyhton解释器的特性，其他类型的解释器如JPyhon，pypy等没有这个特性。之所以我们在使用Python多线程时没有感觉是单线程是因为上下文的切换。

在这里不详细多说，关于GIL全局性解释器锁的详细信息，有兴趣可以参考：https://www.cnblogs.com/cjaaron/p/9166538.html

我们最重要的是需要理解，Python多线程适合IO密集型操作，但在计算密集型操作中，多线程甚至没有单线程快。

再来讲互斥锁（也就是线程锁）

　　当多个线程同时修改一个数据的时候，可能会发生无法预估的错误，所以这时候要上锁。在这里叫做互斥锁。

def run():
    lock.acquire() # 获取锁
    global num
    num += 1
    lock.release() # 释放锁
    time.sleep(1)

if __name__ == "__main__":

    lock = threading.Lock() # 生成锁的实例
    num = 0
    obj = []
    start = time.time()
    for i in range(1000):
        t= threading.Thread(target=run)
        t.start()
        obj.append(t)
    for j in obj:
        j.join()
    end = time.time()
    cost = end - start
    print("cost time:",cost)

    print("num:",num)

互斥锁

递归锁

互斥锁之间可以嵌套，但是没有理清逻辑容易造成死锁，无法解开。这里使用递归锁，实现多层上锁和多层解锁

def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num += 1
    lock.release()
    return num

def run2():
    print("grab the second part data")
    lock.acquire()
    global num2
    num2 += 1
    lock.release()
    return num2

def run3():
    lock.acquire()
    res = run1()
    print('--------between run1 and run2-----')
    res2 = run2()
    lock.release()
    print(res, res2)

if __name__ == '__main__':

    num, num2 = 0, 0
    lock = threading.RLock()  # 递归锁，设锁和解锁必须是成对的
    for i in range(10):
        t = threading.Thread(target=run3)
        t.start()

while threading.active_count() != 1:
    print(threading.active_count())
else:
    print('----all threads done---')
    print(num, num2)

递归锁

需要注意的是：上锁和解锁必须是成对的。

信号量

信号量其实也是一种锁，用于限制线程的并发量，即同一时间只允许几个线程运行。

def run(n):
    semaphore.acquire()
    time.sleep(1)
    print("run the thread: {0}\n".format(n))
    semaphore.release()


if __name__ == '__main__':

    semaphore = threading.BoundedSemaphore(5)  # 最多允许5个线程同时运行
    for i in range(20):
        t = threading.Thread(target=run, args=(i+1,))
        t.start()

while threading.active_count() != 1:
    pass  # print threading.active_count()
else:
    print('----all threads done---')

信号量

Timer

创建一个有延迟的线程。

def hello():
    print("hello, world")
 
t = Timer(30.0, hello)
t.start()  #  30秒后执行此线程

Timer

Event

threading.Event()用来实现两个或多个线程之间的交互

有四个方法：

　　　　set() 设置标志位

　　　　clear()清空标志位

　　　　wait()如果已经设置标志位不会阻塞，如果没有设置标志位则产生阻塞

　　　　isSet() 判定是否设置了标志位。

接下来是一个车等红绿灯的例子，车是停还是行需要根据红绿灯来判定。将实例化若干个车（线程）和一个红绿灯（线程），来实现线程间交互

import threading
import time
import random

def light():
    "红绿灯"
    while True:
        count = 1
        event.set() # 标志位设定了，wait就不阻塞 #绿灯状态
        while True:
            if count == 1:
                print('\033[42;1m--green light on---\033[0m')
                time.sleep(10)
                count = 2
            elif count == 2:
                print('\033[43;1m--yellow light on---\033[0m')
                time.sleep(2)
                count = 3
            elif count == 3:
                event.clear() # 标志位被清空，wait()阻塞
                print('\033[41;1m--red light on---\033[0m')
                time.sleep(10)
                break

def car(n):
    while 1:
        time.sleep(random.randrange(8))
        if  event.isSet(): # 判断标志位是否设定
            print("car [%s] is running.." % n)
        else:
            print("car [%s] is waiting for the red light.." %n)

if __name__ == '__main__':
    event = threading.Event()
    Light = threading.Thread(target=light)
    Light.start()
    for i in range(3):
        t = threading.Thread(target=car,args=(i,))
        t.start()

小车等红绿灯

第二个例子，员工过门。如果门是开的，员工直接过门；如果们是关的，员工需要刷卡开门再过门。

import threading
import time
import random

def door():
    count = 0
    while True:
        if door_event.is_set():
            print("\033[32;1mdoor opening....\033[0m")
            count +=1
        else:
            print("\033[31;1mdoor closed...., swipe to open.\033[0m")
            count = 0 #清空计时器
            door_event.wait()

        if count > 3:#门开了已经3s了,该关了
            door_event.clear()

        time.sleep(0.5)

def staff(n):

    print("staff [%s] is comming..." % n )
    while True:
        if door_event.is_set():
            print("\033[34;1mdoor is opened, passing.....\033[0m")
            break
        else:
            print("staff [%s] sees door got closed, swipping the card....." % n)
            door_event.set()
            print("door is opening...")
        time.sleep(0.5)

if __name__ == "__main__":

    door_event  = threading.Event() #设置事件
    door_thread = threading.Thread(target=door) # 实例化“门”线程并启动
    door_thread.start()

    for i in range(5):
        p = threading.Thread(target=staff,args=(i,))
        time.sleep(random.randrange(3))
        p.start()

员工过门

队列queue

线程间的数据通信。需要导入模块queue。

分为三种：

1. queue.Queue(maxsize=0) 先入先出

　　　qsize() 查询队列中数据的个数

　　　put(item,block=Ture,timeout=None) 往队列中放入数据

　　　get(block=True,timeout=None) 取出数据。当队列中没有数据的时候，get将会造成阻塞，直到队列中又存入数据为止。参数block=False不会阻塞，timeout=1表示只卡1s

　　　empty() 当队列为空返回True

　　　full() 当队列已满返回True

def put():
    #time.sleep(1) # 将执行此函数的线程滞后，更好地观摩多线程的运行效果
    try:
        for i in range(50):
            cou = i
            q.put(i + 1, block=False)
        else:
            print("存入了{0}个".format(cou+1))
    except queue.Full:
        print("队列已满，存入了{0}个".format(cou))


def get():
    try:
        for i in range(50):
            print(q.get(block=False))
    except queue.Empty:
        print("队列已空，无法取出数据")

def qsize():
    # time.sleep(1)
    print("此时队列中个数:",q.qsize())

put_thr = threading.Thread(target=put)
get_thr = threading.Thread(target=get)
qsi_thr = threading.Thread(target=qsize)
q = queue.Queue(5) # 队列最多同时容纳5个数据

put_thr.start()
qsi_thr.start()
get_thr.start()

Queue

在代码中，有用到异常处理，是因为如果队列满了再存入数据会触发异常，队列空了再取数据也会触发异常。

2. queue.LifoQueue(maxsize=0) 先入后出

　　方法与上同

3. PriorityQueue() 存储数据时可设置优先级

import queue
import threading

'''
按照优先级取出数据。
'''

q = queue.PriorityQueue() # 生成队列

q.put((5,"lwz"))
q.put((10,"alex"))
q.put((2,"chenronghua")) # 数字越小优先级越高

print(q.get())
print(q.get())
print(q.get())

PriorityQueue

生产者消费者模型

该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。

为什么要使用生产者和消费者模式?

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

什么是生产者消费者模式?

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

import queue
import threading
import time

def Producer():
    count = 0
    while True:
        count += 1
        q.put("骨头{0}".format(count))
        print("生产了骨头{0}".format(count))
        time.sleep(1)

def Customer(name):
    while 1:
        print("{0}取到并且吃了{1}".format(name,q.get()))
        time.sleep(0.1)


if __name__ == "__main__":

    q = queue.Queue()
    c1 = threading.Thread(target=Customer,args=("alex",))
    c2 = threading.Thread(target=Customer,args=("Chen",))
    p = threading.Thread(target=Producer)

    c1.start()
    c2.start()
    p.start()

生产者消费者模型

微信关注我们

原文链接：https://yq.aliyun.com/articles/621035

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

有Python基础，刚接触web框架的Django初学者

本文面向：有Python基础，刚接触web框架的Django初学者。环境：windows7 python3.5.1 pycharm Django 1.10版 pip3 一、Django简介百度百科：一个开放源代码的Web框架，由Python语言编写...... 重点：一个大而全的框架，啥都替你考虑好了。 1. web框架介绍具体介绍Django之前，必须先介绍WEB框架的概念。 web框架：别人已经设定好的一个web网站模板，你学习它的规则，然后“填空”或“修改”成你自己需要的样子。一般web框架的架构是这样的：其它基于Python的web框架，如tornado、flask、webpy都是在这个范围内进行增删裁剪的。例如Tornado用的是自己的异步非阻塞“wsgi”，flask则只提供了最精简和基本的框架。Django直接使用WSGI，并实现了大部分Web应用相关的功能。我在文末左侧有为你准备Python flask的零基础系统学习课程，点击：了解更多就可以看见了，希望大家多多支持！ 2. MVC/MTV介绍 MVC百度百科：全名Model View Controlle...

2018-07-28

525

JSch连接Sftp使用说明

前言最近在公司需要通过SSH然后对SFTP连接，然后进行操作，所以在这里记录一下，方便自己以后使用，也方便给大家作为参考。正文首先得需要对JSch进行依赖引用，项目由gradle进行搭建。以下是build.gradle 代码： // https://mvnrepository.com/artifact/com.jcraft/jsch compile group: 'com.jcraft', name: 'jsch', version: '0.1.54' 然后需要一个实体类，用来配置我们的JSch的相关参数。 Tips:公司使用的是AWS，参数都是配置在ECS里的。所以我需要从环境变量里取出相应的值，大家可以自行修改成 hard code。实体类： public class SftpConfigProperties { public static final String environment=System.getenv("PUBLISHER_ENVIRONMENT"); public static final String sftpUsername=System.geten...

2018-07-28

817

发表评论

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。