Python之线程

import time
from threading import Thread
#多线程并发，是不是看着和多进程很类似
def func(n):
    time.sleep(1)
    print(n)

#并发效果，1秒打印出了所有的数字
for i in range(10):
    t = Thread(target=func,args=(i,))
    t.start()

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print(‘%s say hello‘ %name)

if __name__ == ‘__main__‘:
    t=Thread(target=sayhi,args=(‘太白‘,))
    t.start()
    print(‘主线程‘)

import time
from threading import Thread
class Sayhi(Thread):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        time.sleep(2)
        print(‘%s say hello‘ % self.name)


if __name__ == ‘__main__‘:
    t = Sayhi(‘太白‘)
    t.start()
    print(‘主线程‘)

from threading import Thread
from multiprocessing import Process
import os

def work():
    print(‘hello‘,os.getpid())

if __name__ == ‘__main__‘:
    #part1:在主进程下开启多个线程,每个线程都跟主进程的pid一样
    t1=Thread(target=work)
    t2=Thread(target=work)
    t1.start()
    t2.start()
    print(‘主线程/主进程pid‘,os.getpid())

    #part2:开多个进程,每个进程都有不同的pid
    p1=Process(target=work)
    p2=Process(target=work)
    p1.start()
    p2.start()
    print(‘主线程/主进程pid‘,os.getpid())

from threading import Thread
from multiprocessing import Process
import os
import time
def work():
    print(‘hello‘)

if __name__ == ‘__main__‘:
    s1 = time.time()
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()
    t.join()
    t1 = time.time() - s1
    print(‘进程的执行时间：‘,t1)
    print(‘主线程/主进程‘)
    ‘‘‘
    打印结果:
    hello
    进程的执行时间： 0.0
    主线程/主进程
    ‘‘‘

    s2 = time.time()
    #在主进程下开启子进程
    t=Process(target=work)
    t.start()
    t.join()
    t2 = time.time() - s2
    print(‘线程的执行时间：‘, t2)
    print(‘主线程/主进程‘)
    ‘‘‘
    打印结果:
    hello
    线程的执行时间： 0.5216977596282959
    主线程/主进程
    ‘‘‘

from  threading import Thread
from multiprocessing import Process
import os
def work():
    global n  #修改全局变量的值
    n=0

if __name__ == ‘__main__‘:
    # n=100
    # p=Process(target=work)
    # p.start()
    # p.join()
    # print(‘主‘,n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100


    n=1
    t=Thread(target=work)
    t.start()
    t.join()   #必须加join，因为主线程和子线程不一定谁快，一般都是主线程快一些，所有我们要等子线程执行完毕才能看出效果
    print(‘主‘,n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据
# 通过一个global就实现了全局变量的使用，不需要进程的IPC通信方法

import multiprocessing
import threading

import socket
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind((‘127.0.0.1‘,8080))
s.listen(5)

def action(conn):
    while True:
        data=conn.recv(1024)
        print(data)
        msg = input(‘服务端输入:‘) #在多线程里面可以使用input输入内容，那么就可以实现客户端和服务端的聊天了，多进程不能输入
        conn.send(bytes(msg,encoding=‘utf-8‘))

if __name__ == ‘__main__‘:

    while True:
        conn,addr=s.accept()
        p=threading.Thread(target=action,args=(conn,))
        p.start()

mport socket

s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect((‘127.0.0.1‘,8080))

while True:
    msg=input(‘>>: ‘).strip()
    if not msg:continue

    s.send(msg.encode(‘utf-8‘))
    data=s.recv(1024)
    print(data)

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

threading模块提供的一些方法：
  # threading.currentThread(): 返回当前的线程变量。
  # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果

from threading import Thread
import threading
from multiprocessing import Process
import os

def work():
    import time
    time.sleep(3)
    print(threading.current_thread().getName())

if __name__ == ‘__main__‘:
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()

    print(threading.current_thread())#主线程对象
    print(threading.current_thread().getName()) #主线程名称
    print(threading.current_thread().ident) #主线程ID
    print(threading.get_ident()) #主线程ID
    print(threading.enumerate()) #连同主线程在内有两个运行的线程
    print(threading.active_count())
    print(‘主线程/主进程‘)

    ‘‘‘
    打印结果:
    <_MainThread(MainThread, started 14104)>
    MainThread
    14104
    14104
    [<_MainThread(MainThread, started 14104)>, <Thread(Thread-1, started 17976)>]
    2
    主线程/主进程
    Thread-1
    ‘‘‘

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print(‘%s say hello‘ %name)

if __name__ == ‘__main__‘:
    t=Thread(target=sayhi,args=(‘太白‘,))
    t2=Thread(target=sayhi,args=(‘alex‘,))
    t.start()
    t2.start()
    t.join()  #因为这个线程用了join方法，主线程等待子线程的运行结束

    print(‘主线程‘)
    print(t.is_alive())  #所以t这个线程肯定是执行结束了，结果为False
    print(t2.is_alive()) #有可能是True，有可能是False，看子线程和主线程谁执行的快
    ‘‘‘
    egon say hello
    主线程
    False
    ‘‘‘

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print(‘%s say hello‘ %name)

if __name__ == ‘__main__‘:
    t=Thread(target=sayhi,args=(‘taibai‘,))
    t.setDaemon(True) #必须在t.start()之前设置
    t.start()

    print(‘主线程‘)
    print(t.is_alive())
    ‘‘‘
    主线程
    True
    ‘‘‘

from threading import Thread
from multiprocessing import Process
import time
def func1():
    while True:
        print(666)
        time.sleep(0.5)
def func2():
    print(‘hello‘)
    time.sleep(3)

if __name__ == ‘__main__‘:
    # t = Thread(target=func1,)
    # t.daemon = True  #主线程结束，守护线程随之结束
    # # t.setDaemon(True) #两种方式，和上面设置守护线程是一样的
    # t.start()
    # t2 = Thread(target=func2,) #这个子线程要执行3秒，主线程的代码虽然执行完了，但是一直等着子线程的任务执行完毕，主线程才算完毕，因为通过结果你会发现我主线程虽然代码执行完毕了，\
    # 但是主线程的的守护线程t1还在执行，说明什么，说明我的主线程还没有完毕，只不过是代码执行完了，一直等着子线程t2执行完毕，我主线程的守护线程才停止，说明子线程执行完毕之后，我的主线程才执行完毕
    # t2.start()
    # print(‘主线程代码执行完啦！‘)
    p = Process(target=func1,)
    p.daemon = True
    p.start()

    p2 = Process(target=func2,)
    p2.start()
    time.sleep(1) #让主进程等1秒，为了能看到func1的打印效果
    print(‘主进程代码执行完啦！‘) #通过结果你会发现，如果主进程的代码运行完毕了，那么主进程就结束了，因为主进程的守护进程p随着主进程的代码结束而结束了，守护进程被回收了，这和线程是不一样的，主线程的代码完了并不代表主线程运行完毕了，需要等着所有其他的非守护的子线程执行完毕才算完毕

from threading import Thread,Lock
import os,time
def work():
    global n
    # lock.acquire() #加锁
    temp=n
    time.sleep(0.1) #一会将下面循环的数据加大并且这里的时间改的更小试试
    n=temp-1
    # time.sleep(0.02)
    # n = n - 1
    ‘‘‘如果这样写的话看不出来效果，因为这样写就相当于直接将n的指向改了，就好比从10，经过1次减1之后，n就直接指向了9，速度太快，看不出效果，那么我们怎么办呢，找一个中间变量来接收n，然后对这个中间变量进行修改，然后再赋值给n，多一个给n赋值的过程，那么在这个过程中间，我们加上一点阻塞时间，来看效果，就像读文件修改数据之后再写回文件的过程。那么这个程序就会出现结果为9的情况，首先一个进程的全局变量对于所有线程是共享的，由于我们在程序给中间变量赋值，然后给n再次赋值的过程中我们加了一些I/O时间，遇到I/O就切换，那么每个线程都拿到了10，并对10减1了，然后大家都得到了9，然后再赋值给n，所有n等于了9‘‘‘
    # lock.release()
if __name__ == ‘__main__‘:
    lock=Lock()
    n=100
    l=[]
    # for i in range(10000):  #如果这里变成了10000，你在运行一下看看结果
    for i in range(100):  #如果这里变成了10000，你在运行一下看看结果
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全

from threading import Thread,Lock
import os,time
def work():
    global n
    lock.acquire() #加锁
    temp=n
    time.sleep(0.1)
    n=temp-1
    lock.release()
if __name__ == ‘__main__‘:
    lock=Lock()
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全

 #1.100个线程去抢GIL锁，即抢执行权限
    #2. 肯定有一个线程先抢到GIL（暂且称为线程1），然后开始执行，一旦执行就会拿到lock.acquire()
    #3. 极有可能线程1还未运行完毕，就有另外一个线程2抢到GIL，然后开始运行，但线程2发现互斥锁lock还未被线程1释放，于是阻塞，被迫交出执行权限，即释放GIL
    #4.直到线程1重新抢到GIL，开始从上次暂停的位置继续执行，直到正常释放互斥锁lock，然后其他的线程再重复2 3 4的过程 

#不加锁:并发执行,速度快,数据不安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    global n
    print(‘%s is running‘ %current_thread().getName())
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == ‘__main__‘:
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

    stop_time=time.time()
    print(‘主:%s n:%s‘ %(stop_time-start_time,n))

‘‘‘
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:0.5216062068939209 n:99
‘‘‘


#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    #未加锁的代码并发运行
    time.sleep(3)
    print(‘%s start to run‘ %current_thread().getName())
    global n
    #加锁的代码串行运行
    lock.acquire()
    temp=n
    time.sleep(0.5)
    n=temp-1
    lock.release()

if __name__ == ‘__main__‘:
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    stop_time=time.time()
    print(‘主:%s n:%s‘ %(stop_time-start_time,n))

‘‘‘
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:53.294203758239746 n:0
‘‘‘

#有的同学可能有疑问:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊
#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是
#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的
#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.
from threading import current_thread,Thread,Lock
import os,time
def task():
    time.sleep(3)
    print(‘%s start to run‘ %current_thread().getName())
    global n
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == ‘__main__‘:
    n=100
    lock=Lock()
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        t.start()
        t.join()
    stop_time=time.time()
    print(‘主:%s n:%s‘ %(stop_time-start_time,n))

‘‘‘
Thread-1 start to run
Thread-2 start to run
......
Thread-100 start to run
主:350.6937336921692 n:0 #耗时是多么的恐怖
‘‘‘

from threading import Thread,Lock
import time
mutexA=Lock()
mutexB=Lock()

class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()
    def func1(self):
        mutexA.acquire()
        print(‘\033[41m%s 拿到A锁>>>\033[0m‘ %self.name)
        mutexB.acquire()
        print(‘\033[42m%s 拿到B锁>>>\033[0m‘ %self.name)
        mutexB.release()
        mutexA.release()

    def func2(self):
        mutexB.acquire()  
        print(‘\033[43m%s 拿到B锁???\033[0m‘ %self.name)
        time.sleep(2)
        #分析：当线程1执行完func1，然后执行到这里的时候，拿到了B锁，线程2执行func1的时候拿到了A锁，那么线程2还要继续执行func1里面的代码，再去拿B锁的时候，发现B锁被人拿了，那么就一直等着别人把B锁释放，那么就一直等着，等到线程1的sleep时间用完之后，线程1继续执行func2，需要拿A锁了，但是A锁被线程2拿着呢，还没有释放，因为他在等着B锁被释放，那么这俩人就尴尬了，你拿着我的老A，我拿着你的B，这就尴尬了，俩人就停在了原地

        mutexA.acquire()
        print(‘\033[44m%s 拿到A锁???\033[0m‘ %self.name)
        mutexA.release()

        mutexB.release()

if __name__ == ‘__main__‘:
    for i in range(10):
        t=MyThread()
        t.start()

‘‘‘
Thread-1 拿到A锁>>>
Thread-1 拿到B锁>>>
Thread-1 拿到B锁???
Thread-2 拿到A锁>>>
然后就卡住，死锁了
‘‘‘

import time
from threading import Thread,Lock
noodle_lock = Lock()
fork_lock = Lock()
def eat1(name):
    noodle_lock.acquire()
    print(‘%s 抢到了面条‘%name)
    fork_lock.acquire()
    print(‘%s 抢到了叉子‘%name)
    print(‘%s 吃面‘%name)
    fork_lock.release()
    noodle_lock.release()

def eat2(name):
    fork_lock.acquire()
    print(‘%s 抢到了叉子‘ % name)
    time.sleep(1)
    noodle_lock.acquire()
    print(‘%s 抢到了面条‘ % name)
    print(‘%s 吃面‘ % name)
    noodle_lock.release()
    fork_lock.release()

for name in [‘taibai‘,‘egon‘,‘wulaoban‘]:
    t1 = Thread(target=eat1,args=(name,))
    t2 = Thread(target=eat2,args=(name,))
    t1.start()
    t2.start()

import time
from threading import Thread,RLock
fork_lock = noodle_lock = RLock()
def eat1(name):
    noodle_lock.acquire()
    print(‘%s 抢到了面条‘%name)
    fork_lock.acquire()
    print(‘%s 抢到了叉子‘%name)
    print(‘%s 吃面‘%name)
    fork_lock.release()
    noodle_lock.release()

def eat2(name):
    fork_lock.acquire()
    print(‘%s 抢到了叉子‘ % name)
    time.sleep(1) 
    noodle_lock.acquire()
    print(‘%s 抢到了面条‘ % name)
    print(‘%s 吃面‘ % name)
    noodle_lock.release()
    fork_lock.release()

for name in [‘taibai‘,‘wulaoban‘]:
    t1 = Thread(target=eat1,args=(name,))
    t1.start()
for name in [‘alex‘,‘peiqi‘]:
    t2 = Thread(target=eat2,args=(name,))
    t2.start()

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import threading
import os,time,random
def task(n):
    print(‘%s is runing‘ %threading.get_ident())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == ‘__main__‘:

    executor=ThreadPoolExecutor(max_workers=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
import os
import threading
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

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

def call_back(m):
    print(‘结果为：%s‘%(m.result()))

tpool = ThreadPoolExecutor(max_workers=5)
t_lst = []
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()拿到结果