多线程

第九章

1. 线程
2. python GIL全局解释器锁
3. 线程
1. 语法
2. join
3. 线程锁之LockRlock信号量
4. 将线程变为守护进程
5. Event事件　
6. queue队列
7. 生产者消费者模型
8. Queue队列

线程

一.什么是线程

线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务

一个线程是一个执行上下文，这是所有的信息，一个处理器需要执行一个指令流。

假设你正在读一本书，你想现在休息，但你想能够回来，并恢复阅读从确切的点，你停了。一个实现的方法就是记下页码，行数和字数。因此，你读一本书的执行上下文是这3个数字。

如果你有一个室友，她使用相同的技术，她可以把这本书，而你不使用它，并恢复阅读从她停下来的地方。然后你可以把它回来，从你的地方恢复。

线程以同样的方式工作。一个中央处理器给你的错觉，它同时做多个计算。它通过在每一个计算上花费一点时间。它可以这样做，因为它有一个每个计算的执行上下文。就像你可以和你的朋友分享一本书，许多任务可以共享一个处理器。

在一个更技术的层面上，一个执行上下文（因此一个线程）由中央处理器的寄存器的值组成。

最后：线程与进程不同。线程是一个执行的上下文，而一个进程是一堆与计算相关的资源。一个进程可以有一个或多个线程。

澄清：与进程相关联的资源包括内存页（进程中的所有线程都有相同的内存视图），文件描述符（例如，打开的套接字），和安全凭据（例如，启动进程的用户的身份）。

 

二.什么是过程

一个程序的一个执行实例被称为一个过程。

每个进程提供执行程序所需的资源。一个进程有一个虚拟地址空间，可执行代码，开放的处理系统对象，一个安全上下文，一个独特的过程标识符，环境变量，优先级类，最小和最大工作集大小，和至少一个执行线程。每个进程都是开始于一个单一的线程，通常被称为主线程，但可以创建额外的线程从它的任何线程。

三.线程和进程的区别

1.线程共享创建它的进程的地址空间；进程有自己的地址空间。

2.线程对其进程的数据段有直接的访问；进程有自己的父进程的数据段的副本。

3.线程可以直接与其他线程的过程；过程必须使用进程间通信与兄弟姐妹的过程。

4.新的线程很容易创建；新的进程需要重复的父进程。

5.线程可以对同一进程的线程进行相当程度的控制，过程只能对子进程进行控制。

6.更改主线程（取消、优先级更改等）可能会影响进程的其他线程的行为；父进程的更改不影响子进程。

python GIL全局解释器锁

在CPython，全局解释器锁，或吉尔，是一个互斥体防止多个本地线程执行Python字节码一次。这把锁是必要的主要是因为当前的内存管理不是线程安全的。（然而，由于吉尔的存在，其他功能已经增长，取决于它强制执行的保证。）

首先需要明确的一点是GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。 就好比C++是一套语言（语法）标准，但是可以用不同的编译器来编译成可执行代码。有名的编译器例如GCC，INTEL C++，Visual C++等。Python也一样，同样一段代码可以通过CPython，PyPy，Psyco等不同的Python执行环境来执行。像其中的JPython 就没有GIL。然而因为CPython是大部分环境下默认的Python执行环境。所以在很多人的概念里CPython就是Python，也就想当然的把GIL归结为Python语言的缺陷。所以这里要先明确一点：GIL并不是Python的特性，Python完全可以不依赖于GIL

Python threading模块

调用

直接调用：

import threading
import time
 
def sayhi(num): #定义每个线程要运行的函数
 
    print("running on number:%s" %num)
 
    time.sleep(3)
 
if __name__ == '__main__':
 
    t1 = threading.Thread(target=sayhi,args=(1,)) #生成一个线程实例
    t2 = threading.Thread(target=sayhi,args=(2,)) #生成另一个线程实例
 
    t1.start() #启动线程
    t2.start() #启动另一个线程
 
    print(t1.getName()) #获取线程名
    print(t2.getName())

继承调用：

import threading
import time
 
 
class MyThread(threading.Thread):
    def __init__(self,num):
        threading.Thread.__init__(self)
        self.num = num
 
    def run(self):#定义每个线程要运行的函数
 
        print("running on number:%s" %self.num)
 
        time.sleep(3)
 
if __name__ == '__main__':
 
    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

线程锁(互斥锁Mutex)

一个进程下可以启动多个线程，多个线程共享父进程的内存空间，也就意味着每个线程可以访问同一份数据，此时，如果2个线程同时要修改同一份数据，会出现什么状况？

import time
import threading
 
def addNum():
    global num #在每个线程中都获取这个全局变量
    print('--get num:',num )
    time.sleep(1)
    num  -=1 #对此公共变量进行-1操作
 
num = 100  #设定一个共享变量
thread_list = []
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
 
for t in thread_list: #等待所有线程执行完毕
    t.join()
 
 
print('final num:', num )

正常来讲，这个num结果应该是0， 但在python 2.7上多运行几次，会发现，最后打印出来的num结果不总是0，为什么每次运行的结果不一样呢？ 哈，很简单，假设你有A,B两个线程，此时都 要对num 进行减1操作， 由于2个线程是并发同时运行的，所以2个线程很有可能同时拿走了num=100这个初始变量交给cpu去运算，当A线程去处完的结果是99，但此时B线程 运算完的结果也是99，两个线程同时CPU运算的结果再赋值给num变量后，结果就都是99。那怎么办呢？ 很简单，每个线程在要修改公共数据时，为了避免自己在还没改完的时候别人也来修改此数据，可以给这个数据加一把锁， 这样其它线程想修改此数据时就必须等待你修改完毕并把锁释放掉后才能再访问此数据。 

*注：不要在3.x上运行，不知为什么，3.x上的结果总是正确的，可能是自动加了锁

加锁版本

import time
import threading
 
def addNum():
    global num #在每个线程中都获取这个全局变量
    print('--get num:',num )
    time.sleep(1)
    lock.acquire() #修改数据前加锁
    num  -=1 #对此公共变量进行-1操作
    lock.release() #修改后释放
 
num = 100  #设定一个共享变量
thread_list = []
lock = threading.Lock() #生成全局锁
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
 
for t in thread_list: #等待所有线程执行完毕
    t.join()
 
print('final num:', num )

GIL VS Lock 

机智的同学可能会问到这个问题，就是既然你之前说过了，Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要 lock? 注意啦，这里的lock是用户级的lock,跟那个GIL没关系 ，具体我们通过下图来看一下+配合我现场讲给大家，就明白了。

 

RLock（递归锁）

说白了就是在一个大锁中还要再包含子锁

import threading,time
 
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)

Semaphore(信号量)

互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。

import threading,time
 
def run(n):
    semaphore.acquire()
    time.sleep(1)
    print("run the thread: %s
" %n)
    semaphore.release()
 
if __name__ == '__main__':
 
    num= 0
    semaphore  = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
    for i in range(20):
        t = threading.Thread(target=run,args=(i,))
        t.start()
 
while threading.active_count() != 1:
    pass #print threading.active_count()
else:
    print('----all threads done---')
    print(num)

Events

一个事件是一个简单的同步对象；

该事件表示一个内部标志，和线程

可以等待要设置的标志，或设置或清除标志本身。

事件= event()线程。

#客户端线程可以等待标志被设置

wait()事件。

#服务器线程可以设置或重置

set()事件。

clear()事件。

如果设置了标志，则等待方法不做任何事情。

如果该标志被清除，等待将阻止，直到它再次被设置。

任意数量的线程可以等待相同的事件。

通过Event来实现两个或多个线程间的交互，下面是一个红绿灯的例子，即起动一个线程做交通指挥灯，生成几个线程做车辆，车辆行驶按红灯停，绿灯行的规则。

import threading,time
import random
def light():
    if not event.isSet():
        event.set() #wait就不阻塞 #绿灯状态
    count = 0
    while True:
        if count < 10:
            print('33[42;1m--green light on---33[0m')
        elif count <13:
            print('33[43;1m--yellow light on---33[0m')
        elif count <20:
            if event.isSet():
                event.clear()
            print('33[41;1m--red light on---33[0m')
        else:
            count = 0
            event.set() #打开绿灯
        time.sleep(1)
        count +=1
def car(n):
    while 1:
        time.sleep(random.randrange(10))
        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()

这里还有一个event使用的例子，员工进公司门要刷卡， 我们这里设置一个线程是“门”， 再设置几个线程为“员工”，员工看到门没打开，就刷卡，刷完卡，门开了，员工就可以通过。

#_*_coding:utf-8_*_
__author__ = 'Alex Li'
import threading
import time
import random

def door():
    door_open_time_counter = 0
    while True:
        if door_swiping_event.is_set():
            print("33[32;1mdoor opening....33[0m")
            door_open_time_counter +=1

        else:
            print("33[31;1mdoor closed...., swipe to open.33[0m")
            door_open_time_counter = 0 #清空计时器
            door_swiping_event.wait()


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

        time.sleep(0.5)


def staff(n):

    print("staff [%s] is comming..." % n )
    while True:
        if door_swiping_event.is_set():
            print("33[34;1mdoor is opened, passing.....33[0m")
            break
        else:
            print("staff [%s] sees door got closed, swipping the card....." % n)
            print(door_swiping_event.set())
            door_swiping_event.set()
            print("after set ",door_swiping_event.set())
        time.sleep(0.5)
door_swiping_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队列

当信息必须在多个线程之间安全交换的时候，队列在线程编程是特别有用的。

班队列。队列（大小= 0）#先入先出

队列类。LifoQueue（大小= 0）#最后一出

班队列，优先级队列（大小= 0）#存储数据时可设置优先级的队列

优先级队列的构造函数。为是集上，可以放置在队列中的项的数量极限上限整数。插入将块一旦这个大小已达到，直到队列项目被消耗。如果不能小于或等于零，队列的大小是无限的。

最低值的条目检索第一（最低值的条目是一个返回的排序（列表（条目））[ 0 ]）。作品的典型模式是在形成一个元组：（priority_number，数据）。

异常队列。

异常时引发的非阻塞get()（或get_nowait()）被一个队列对象是空的。

异常队列

异常时引发的非阻塞put()（或put_nowait()）被一个队列对象是全。

qsize()队列。

队列。empty() #返回true，如果是空的

队列。full() #返回true，如果全

队列。放置（项目，块=真，超时=无）

把项目放在队列中。如果可选的参数块是真实和超时没有（默认），如果有必要，直到块空闲时隙可用。如果超时是一个积极的数字，它块在大多数超时秒，并提出了一个完整的例外，如果没有空闲时隙在该时间内。否则（块是错误的），把一个项目的队列，如果一个空闲时隙立即可用，否则提高了完整的例外（超时被忽略在这种情况下）。

队列。put_nowait（项目）

相当于放（项目，假）。

队列。获取（块=真，超时=无）

从队列中删除并返回一个项目。如果可选的参数块是真实和超时没有（默认），如果有必要，直到项目块是可用的。如果超时是一个积极的数字，它块在大多数超时秒，并提出了空的例外，如果没有项目是在该时间内。否则（块是错误的），返回一个项目，如果一个是立即可用，否则提高空的异常（超时被忽略在这种情况下）。

get_nowait()队列。

相当于得到（假）。

两方法提供给支持跟踪是否入队的任务已经由后台消费者线程完全处理。

task_done()队列。

表明以前排队任务完成。使用队列的消费线程。每个get()用于获取任务，随后打电话告诉task_done()队列的任务处理完成。

如果一个join()目前阻塞，它将恢复在已处理所有的项目（即task_done()收到电话每项已put()插队）。

引发ValueError如果有比放在队列中的项的次数。

join()块被消费完毕直到队列排队。

 

生产者消费者模型

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

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

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

什么是生产者消费者模式

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

下面来学习一个最基本的生产者消费者模型的例子

import threading
import queue

def producer():
    for i in range(10):
        q.put("骨头 %s" % i )

    print("开始等待所有的骨头被取走...")
    q.join()
    print("所有的骨头被取完了...")


def consumer(n):

    while q.qsize() >0:

        print("%s 取到" %n  , q.get())
        q.task_done() #告知这个任务执行完了


q = queue.Queue()



p = threading.Thread(target=producer,)
p.start()

c1 = consumer("陈荣华")

import time,random
import queue,threading
q = queue.Queue()
def Producer(name):
  count = 0
  while count <20:
    time.sleep(random.randrange(3))
    q.put(count)
    print('Producer %s has produced %s baozi..' %(name, count))
    count +=1
def Consumer(name):
  count = 0
  while count <20:
    time.sleep(random.randrange(4))
    if not q.empty():
        data = q.get()
        print(data)
        print('33[32;1mConsumer %s has eat %s baozi...33[0m' %(name, data))
    else:
        print("-----no baozi anymore----")
    count +=1
p1 = threading.Thread(target=Producer, args=('A',))
c1 = threading.Thread(target=Consumer, args=('B',))
p1.start()
c1.start()

Paramiko模块

SSHClient

用于连接远程服务器并执行基本命令

基于用户名密码连接：

import paramiko
  
# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='c1.salt.com', port=22, username='wupeiqi', password='123')
  
# 执行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 获取命令结果
result = stdout.read()
  
# 关闭连接
ssh.close()

import paramiko

transport = paramiko.Transport(('hostname', 22))
transport.connect(username='wupeiqi', password='123')

ssh = paramiko.SSHClient()
ssh._transport = transport

stdin, stdout, stderr = ssh.exec_command('df')
print stdout.read()

transport.close()

基于公钥密钥连接：

import paramiko
 
private_key = paramiko.RSAKey.from_private_key_file('/home/auto/.ssh/id_rsa')
 
# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='c1.salt.com', port=22, username='wupeiqi', key=private_key)
 
# 执行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 获取命令结果
result = stdout.read()
 
# 关闭连接
ssh.close()

import paramiko

private_key = paramiko.RSAKey.from_private_key_file('/home/auto/.ssh/id_rsa')

transport = paramiko.Transport(('hostname', 22))
transport.connect(username='wupeiqi', pkey=private_key)

ssh = paramiko.SSHClient()
ssh._transport = transport

stdin, stdout, stderr = ssh.exec_command('df')

transport.close()

SFTPClient

用于连接远程服务器并执行上传下载

基于用户名密码上传下载

import paramiko
 
transport = paramiko.Transport(('hostname',22))
transport.connect(username='wupeiqi',password='123')
 
sftp = paramiko.SFTPClient.from_transport(transport)
# 将location.py 上传至服务器 /tmp/test.py
sftp.put('/tmp/location.py', '/tmp/test.py')
# 将remove_path 下载到本地 local_path
sftp.get('remove_path', 'local_path')
 
transport.close()

基于公钥密钥上传下载

import paramiko
 
private_key = paramiko.RSAKey.from_private_key_file('/home/auto/.ssh/id_rsa')
 
transport = paramiko.Transport(('hostname', 22))
transport.connect(username='wupeiqi', pkey=private_key )
 
sftp = paramiko.SFTPClient.from_transport(transport)
# 将location.py 上传至服务器 /tmp/test.py
sftp.put('/tmp/location.py', '/tmp/test.py')
# 将remove_path 下载到本地 local_path
sftp.get('remove_path', 'local_path')
 
transport.close(

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import paramiko
import uuid

class Haproxy(object):

    def __init__(self):
        self.host = '172.16.103.191'
        self.port = 22
        self.username = 'wupeiqi'
        self.pwd = '123'
        self.__k = None

    def create_file(self):
        file_name = str(uuid.uuid4())
        with open(file_name,'w') as f:
            f.write('sb')
        return file_name

    def run(self):
        self.connect()
        self.upload()
        self.rename()
        self.close()

    def connect(self):
        transport = paramiko.Transport((self.host,self.port))
        transport.connect(username=self.username,password=self.pwd)
        self.__transport = transport

    def close(self):

        self.__transport.close()

    def upload(self):
        # 连接，上传
        file_name = self.create_file()

        sftp = paramiko.SFTPClient.from_transport(self.__transport)
        # 将location.py 上传至服务器 /tmp/test.py
        sftp.put(file_name, '/home/wupeiqi/tttttttttttt.py')

    def rename(self):

        ssh = paramiko.SSHClient()
        ssh._transport = self.__transport
        # 执行命令
        stdin, stdout, stderr = ssh.exec_command('mv /home/wupeiqi/tttttttttttt.py /home/wupeiqi/ooooooooo.py')
        # 获取命令结果
        result = stdout.read()


ha = Haproxy()
ha.run()