线程、锁

进程与线程

　　进程：以一个整体的形式暴露给操作系统管理，里面包含各种资源的调用，是各种资源的集合（qq）

　　线程：操作系统最小的调度单位，是一串指令的集合，进程要操作 CPU，必须要先创建一个线程 

线程与进程的区别：

1. 线程共享内存空间，进程的内存是独立的
2  
2. 同一个进程的线程之间可以直接交流，两个进程想通信，必须通过一个中间代理来实现

3. 创建新线程很简单， 创建新进程需要对其父进程进行一次克隆

4. 一个线程可以控制和操作同一进程里的其他线程，但是进程只能操作子进程

import threading
import time


def run(n):
    time.sleep(2)
    print("run:",n)

t1 = threading.Thread(target=run,args=("t1",))  # 创建一个线程，args必须是元组

t2 = threading.Thread(target=run,args=("t2",))   # 创建另一个线程

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

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

    def run(self):  # 必须是run，方法定义为其他名称不执行
        time.sleep(1)
        print("run:",self.n)

t1 = MyThread("t1")
t2 = MyThread("t2")

t1.start()
t2.start()

t.join()：等待某个线程执行完成

import threading
import time

def run(n):
    print("run:",n)
    time.sleep(2)

t_list = []
start_time = time.time()
for i in range(20):
    t = threading.Thread(target=run,args=("t-%s" % i,))  # 生成一个线程
    t.start()
    t_list.append(t)   # 生成的每个线程加入列表

print(t_list)

for x in t_list:
    x.join()   # 等待每个线程执行结束


# 如果不进行join等待线程，主线程会直接执行下去，不会等其他子线程执行完
print("----- cost:",time.time() - start_time)
print("main threading ....")

t.setDaemon()：守护线程，非守护线程（主线程）执行完就直接退出，不会管守护线程是否执行完成(运用例子：socket多线程连接stop后，守护线程也都全部退出)

import threading
import time


def run(n):
    print("run:",n,threading.current_thread())  # threading.current_thread() 查看当前线程
    time.sleep(2)

start_time = time.time()
for i in range(30):
    t = threading.Thread(target=run,args=("t-%i" % i,))
    t.setDaemon(True)   # 将线程设置为守护线程,设置要在start执行前
    t.start()

print("-----cost-----:",time.time() - start_time,threading.current_thread())

GIL锁、线程锁

GIL锁简单介绍：

　　无论你启多少个线程，你有多少个cpu, Python在执行的时候会淡定的在同一时刻只允许一个线程运行。

　　GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念，CPython是大部分环境下默认的Python执行环境，JPython中没有GIL。

　　线程执行的时候会先申请一个GIL锁

GIL锁问题：

　　线程修改数据时，第一个线程没执行完,GIL锁就被要求释放,线程记录好上下文后对GIL进行释放。恰好另一个线程也要修改同一个数据,执行完后返回了数据。第一个线程再次执行时申请GIL锁进行运算时不会在查数据,会从上次执行到的地方继续往后执行。那样就会出现第一个线程跟另一个线程执行的是同一个结果,并且覆盖,会出现数据不准确的情况

解决方法：线程锁，在加一把锁把数据锁上，当一个线程在执行数据修改时，没有执行完，其他线程不能修改数据

线程锁（互斥锁）：是用户态的锁,可以自己先把要修改的数据锁上,那样其他线程就改不了,等执行完释放锁后其他线程才可以执行

def run():
    global num
    lock.acquire()  # 加线程锁
    num += 1
    lock.release()   # 释放线程锁
    time.sleep(2)


num = 0
lock = threading.Lock()   # 线程锁
t_obj = []
for i in range(50):
    t = threading.Thread(target=run,args=())
    t.start()
    t_obj.append(t)


for t in t_obj:   # 等待所有线程执行完
    t.join()

print("-----Main Threading-----")
print("num:",num)

加入GIL主要的原因是为了降低程序的开发的复杂度，比如现在的你写python不需要关心内存回收的问题，因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的，此时你自己的程序 里的线程和 py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了，为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，这样就解决了上述的问题，  这可以说是Python早期版本的遗留问题。

递归锁

递归锁：同时用到多次锁的情况下一定要用递规锁,不然会被锁死退不出来,递规锁相当于每加一把锁,记录下是哪把锁,开锁的时候根据记录的情况去开锁

import threading


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

# run3中不管怎么运行,都会出现有两把锁的情况
def run3():
    lock.acquire()
    res = run1()  # run1中有加锁
    print("------ between run1 and run2 ----------")
    res2 = run2()  # run2中有加锁
    lock.release()
    print(res,res2)

if __name__ == "__main__":
    num,num2 = 0,0
    #lock = threading.Lock()   # 线程锁,多把锁同时出现时,会出现死锁的情况
    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
import time

def run(n):
    semaphore.acquire()
    print("run the thread: %s
" % n)
    global num
    num += 1
    time.sleep(1)
    semaphore.release()


if __name__ == "__main__":
    num = 0
    semaphore = threading.BoundedSemaphore(5)  # 最多允许5个线程同时运行
    for i in range(22):
        t = threading.Thread(target=run,args=(i,))
        t.start()

while threading.active_count() != 1:
    # time.sleep(0.5)
    print("当前活跃线程数：",threading.active_count())
else:
    print("---- all threads done ----")
    print("num:",num)

Event事件

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

event = threading.Event()   # 生成一个事件　
event.wait() # 当标识位没有设置时,一直等待直到标识位设定
event.set() # 设定标识位
event.clear() # 清除标识位
event.isSet()  # 判断是否设置了标识位

import threading
import time
import random


def light():
    if not event.isSet():  # 判断事件标识位是否被设置
        event.set()   # 设置标识位,wait不阻塞
    count = 1
    while True:
        if count < 5:
            print("33[42;1m green light on....33[0m")   # 绿灯
        elif count < 8:
            print("33[41;1m red light on ....33[0m")   # 红灯
            event.clear()   # 清空标识位,wait进行等待
        else:
            event.set()  # 重设置标识位
            count = 0
        time.sleep(1)
        count += 1


def cars(n):
    while True:
        time.sleep(random.randrange(10))  # 随机数
        if event.isSet():
            print("car [%s] is running..." %n)
        else:
            print("car [%s] is waiting for 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=cars,args=(i,))
        t.start()

queue队列

class queue.Queue(maxsize=0) #先入先出

class queue.LifoQueue(maxsize=0) #last in fisrt out 

class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列，插入的格式：元组(priority_number, data)，按元组的第一个值排序

exception queue.Empty   # 队列为空：get()设置了超时时间、get_nowait()就会报这错

exception queue.Full   # 队列满了：put()、put_nowait() 会报这错

Queue.qsize()   # 队列长度

Queue.empty() #return True if empty  

Queue.full() # return True if full 

Queue.put(item, block=True, timeout=None)  # block 阻塞

    Put item into the queue. If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Full exception if no free slot was available within that time. Otherwise (block is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (timeout is ignored in that case).

Queue.put_nowait(item)

    Equivalent to put(item, False).

Queue.get(block=True, timeout=None)

    Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Empty exception if no item was available within that time. Otherwise (block is false), return an item if one is immediately available, else raise the Empty exception (timeout is ignored in that case).

Queue.get_nowait()

    Equivalent to get(False).

Two methods are offered to support tracking whether enqueued tasks have been fully processed by daemon consumer threads.

Queue.task_done()

    Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.

    If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue).

    Raises a ValueError if called more times than there were items placed in the queue.

Queue.join() # block直到queue被消费完毕

import time
import queue
import threading


def producer():
    while True:
        for i in range(10):
            q.put("骨头 %s " % i)  # 放数据到队列中
        print("开始等待所有骨头被取走....")
        q.join()   # 等待队列中所有数据被取完
        print("所有骨头被取完了。。。")


def consumer(n):
    while True:
        if q.qsize() == 0:   # 队列大小
            print("33[41;1m没有了,等待生产者33[0m")
            time.sleep(1)
        print("%s 取到" %n, q.get())  # 取队列中的数据
        q.task_done()  # 告知这个任务执行完成

# 队列
# q = queue.Queue()   # 先进先出
q = queue.LifoQueue()   # 后进先出

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

c1 = consumer("李闯")

import queue

q = queue.PriorityQueue()   # 优先级队列

q.put((0,"alex"))
q.put((-3,"sb"))
q.put((2,"hehe"))

while q.qsize() > 0:
    print(q.get())