Python 线程----线程方法,线程事件,线程队列,线程池,GIL锁,协程,Greenlet

from threading import Thread
import threading
import time

def work():
    time.sleep(1)
    print("子线程对象>>>", threading.current_thread())               # 子线程对象
    print("子线程名称>>>", threading.current_thread().getName())     # 子线程名称
    print("子线程ID>>>", threading.get_ident())                      # 子线程ID

if __name__ == ‘__main__‘:
    t = Thread(target=work)     # 创建子线程
    t.start()                   # 开启子线程
    print("主线程对象>>>", threading.current_thread())           # 主线程对象
    print("主线程名称>>>", threading.current_thread().getName()) # 主线程名称
    print("主线程ID>>>", threading.current_thread().ident)       # 主线程ID
    print("主线程ID>>>", threading.get_ident())                  # 主线程ID

    time.sleep(1)                   # 阻塞住,此时主线程代码运行的同时子线程代码也在运行
    print(threading.enumerate())    # 拿到所有正在运行的线程对象(包括主线程)
    print(threading.active_count()) # 拿到所有正在运行的线程对象的数量
    print("主线程/主进程执行完毕")

from threading import Thread,Event
import threading
import time,random
def conn_mysql():
    count=1
    while not event.is_set():
        if count > 3:
            raise TimeoutError(‘链接超时‘) #自己发起错误
        print(‘<%s>第%s次尝试链接‘ % (threading.current_thread().getName(), count))
        event.wait(0.5) #
        count+=1
    print(‘<%s>链接成功‘ %threading.current_thread().getName())


def check_mysql():
    print(‘\033[45m[%s]正在检查mysql\033[0m‘ % threading.current_thread().getName())
    t1 = random.randint(0,3)
    print(‘>>>>‘,t1)
    time.sleep(t1)
    event.set()
if __name__ == ‘__main__‘:
    event=Event()
    check = Thread(target=check_mysql)
    conn1=Thread(target=conn_mysql)
    conn2=Thread(target=conn_mysql)

    check.start()
    conn1.start()
    conn2.start()

import queue #不需要通过threading模块里面导入，直接import queue就可以了，这是python自带的,用法基本和我们进程multiprocess中的queue是一样的

q=queue.Queue()
q.put(‘first‘)
q.put(‘second‘)
q.put(‘third‘)
# q.put_nowait() #没有数据就报错，可以通过try来搞

print(q.get())
print(q.get())
print(q.get())
# q.get_nowait() #没有数据就报错，可以通过try来搞

# 执行结果: (先进先出)
# first
# second
# third

import queue

q=queue.LifoQueue() #队列，类似于栈，后进先出的顺序
q.put(‘first‘)
q.put(‘second‘)
q.put(‘third‘)
# q.put_nowait()

print(q.get())
print(q.get())
print(q.get())
# q.get_nowait()

# 执行结果:(后进先出)
# third
# second
# first

rt queue

q = queue.PriorityQueue()   # 创建栈

# put()方法放置一个元组进入栈,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((-10, "a"))
q.put((-5, "a"))  # 负数也可以
# q.put((20,"ws"))  # 如果两个值的优先级一样,那么按照后面的值的acsii码顺序来排序,如果字符串第一个数元素相同,比较第二个元素的acsii码顺序
# q.put((20,"wd"))
# q.put((20,{"a": 11})) # TypeError: unorderable types: dict() < dict() 不能是字典
# q.put((20,("w", 1)))  # 优先级相同的两个数据,他们后面的值必须是相同的数据类型才能比较,可以是元祖,也是通过元素的ascii码顺序来排序

q.put((20, "b"))
q.put((20, "a"))
q.put((0, "b"))
q.put((30, "c"))

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

# 数字越小优先级越高,优先级高的优先出队

import time
from concurrent.futures import ThreadPoolExecutor

def func(n):
    time.sleep(2)
    return n**2

if __name__ == ‘__main__‘:
    thread_pool = ThreadPoolExecutor(max_workers=4) # 创建线程池对象,默认一般开启的线程数量不超过CPU个数的5倍
    t_list = []
    for i in range(20):
        t = thread_pool.submit(func, i) # 提交执行函数,返回一个结果对象,i作为任务函数的参数.submit(func, *args, **kwargs)可以传递任意形式的参数
        t_list.append(t)

    # thread_pool.shutdown()  # shutdown()的作用相当于close()和join(),等待所有的线程执行完毕
    # for tt in t_list:
    #     print(">>>", tt.result())

    # 也可以不用shutdown()方法,改换下面这种方式:
    for n, tt in enumerate(t_list):   # enumerate()枚举
        print(">>>", n, tt.result())
    time.sleep(2)

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import threading
import time, random

def task(n):
    print("{} is runing".format(threading.get_ident())) # 子线程ID号
    time.sleep(random.randint(1, 3))
    return n**2

if __name__ == ‘__main__‘:
    executor = ThreadPoolExecutor(max_workers=3)    # 创建线程池对象,设置的线程数量为3
    for i in range(11):
        future = executor.submit(task, i)
    s = executor.map(task, range(1, 5)) # map()取代了 for循环+submit()
    print([i for i in s])

import time
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

def func(n):
    time.sleep(2)
    return n*n

def call_back(m):
    print("结果为:{}".format(m.result()))

if __name__ == ‘__main__‘:
    tpool = ThreadPoolExecutor(max_workers=5)   # 创建进程池对象
    t_list = []
    for i in range(5):
        t = tpool.submit(func, i).add_done_callback(call_back)

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os

def get_page(url):
    print(‘<进程%s> get %s‘ %(os.getpid(),url))
    respone=requests.get(url)
    if respone.status_code == 200:
        return {‘url‘:url,‘text‘:respone.text}

def parse_page(res):
    res=res.result()
    print(‘<进程%s> parse %s‘ %(os.getpid(),res[‘url‘]))
    parse_res=‘url:<%s> size:[%s]\n‘ %(res[‘url‘],len(res[‘text‘]))
    with open(‘db.txt‘,‘a‘) as f:
        f.write(parse_res)


if __name__ == ‘__main__‘:
    urls=[
        ‘https://www.baidu.com‘,
        ‘https://www.python.org‘,
        ‘https://www.openstack.org‘,
        ‘https://help.github.com/‘,
        ‘http://www.sina.com.cn/‘
    ]

    # p=Pool(3)
    # for url in urls:
    #     p.apply_async(get_page,args=(url,),callback=pasrse_page)
    # p.close()
    # p.join()

    p=ProcessPoolExecutor(3)
    for url in urls:
        p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果