Python多线程进程入门1

Posted 2020-08-23

进程是资源的一个集合,

1、一个应用程序，可以有多进程和多线程

2、默认一个程序是单进程单线程

IO操作使用多线程提高并发

计算操作使用多进程提高并发

进程与线程区别

1、线程共享内存空间，进程的内存是独立的

2、线程共享进程内存的数据，进程之间的数据是无法进行访问的

3、在同一个进程内，线程之间可以直接进行数据的访问，两个进程如果通信，必须通过一个中间代理进行通信

4、创建线程非常容易，创建新进程需要对其父进程进行一次克隆

5、一个线程可以控制和操作同一进程里的其它线程，但是进程只能操作子进程

 

 一个简单的创建多线程例子:

import threading
import time

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


# 多线程
t1 = threading.Thread(target=run, args=("t1",)) # 要创建一个线程，并让线程执行run方法
t2 = threading.Thread(target=run, args=("t2",))
t1.start()  # 并不代表当前线程会被立即执行，需要等待cpu进行调度
t2.start()  # 启动另一个线程

print(t1.getName())  # 获取线程名
print(t2.getName())

# 非多线程
# run(‘t1‘)
# run(‘t2‘)

利用for循环创建多线程:

import threading
import time

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

for i in range(50):
    t = threading.Thread(target=run, args=("t-%s" % i,))
    t.start()

那主线程是否等子线程呢？举个例子

通过这个例子可以看出，当主线程执行完毕后，等待子线程

import time
import threading

def run(num):
    time.sleep(5)
    print(num)

t1 = threading.Thread(target=run,args=(1,))
t1.start()
print(‘end‘)

setDaemon方法:主线程执行完毕后，不等子线程，所以在这里你永远看不见子线程有结果

import time
import threading


def run(num):
    time.sleep(5)
    print(num)


for i in range(10):
    t1 = threading.Thread(target=run, args=(i,))
    t1.setDaemon(True)  # 设置成守护线程，true ,表示主线程不等子线程
    t1.start()
print(‘end‘)

 join方法:等待子线程执行完毕后，再继续

import time
import threading


def run(num):
    time.sleep(2)
    print(num)


for i in range(5):
    t1 = threading.Thread(target=run, args=(i,))
    t1.start()
    t1.join()  # 主线程执行到这里就等待，直到子线程执行完毕后，再继续
print(‘end‘)

上边的例子貌似解决了，等待子线程都执行完毕后，主线程继续执行，但是变为串行了，怎么解决呢？

看下面的例子

import time
import threading


def run(num):
    time.sleep(2)
    print(num)

t_list = []
for i in range(5):
    t1 = threading.Thread(target=run, args=(i,))
    t1.start()
    t_list.append(t1)

for t in t_list:
    t.join()

print(‘main thread‘)

 

join()+参数:表示主线程在此最多等N秒

import time
import threading


def run(num):
    time.sleep(2)
    print(num)


for i in range(5):
    t1 = threading.Thread(target=run, args=(i,))
    # t1.setDaemon(True)  # true ,表示主线程不等子线程
    t1.start()
    t1.join(2)  # 主线程执行到这里就等待，直到子线程执行完毕后，再继续
print(‘end‘)

 

import threading


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

    def run(self):
        self.func(self.args)


def f2(arg):
    print(arg)


obj = MyThread(f2, 1234)
obj.start()

 

import threading


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

    def run(self):  # 必须是run名字，定义每个线程要运行的函数
        print("running task", self.n)


t1 = MyThread(‘t1‘)
t2 = MyThread(‘t2‘)

t1.start()
t2.start()

queue模块

import queue

q = queue.Queue(maxsize=2)  # 默认先进先出,可以添加参数maxsize=2
q1 = queue.LifoQueue()  # last in fisrt out 先进后出
q2 = queue.PriorityQueue  # 优先级

# put 放数据
# get 取数据
# 队列最大长度

q.put(1)
q.put(2)
q.put(3, block=False)  # 当队列默认最大2条消息时候，再放就等待，可以在put里面增加block=False，timeout参数，就不等待了

task_done()和join方法

import queue

q = queue.Queue(5)

q.put(1)
q.put(2)

q.get()
q.task_done()  # 告诉队列这个数据我取出来
q.get()
q.task_done()
q.join()  # 如果队列里面还有数据，我就等待，否则就终止，这个参数需要task_done配合一起是用

优先级队列

q = queue.PriorityQueue()

q.put([3, ‘abc‘])
q.put([0, ‘ccc‘])
print(q.get())

多线程锁机制

防止多个线程同时修改同一个共享数据

例子1:模拟多个线程修改同一个共享数据

import time
import threading

NUM = 10

def func():
    global NUM
    NUM -= 1
    time.sleep(2)
    print(NUM)

t_list = []
for i in range(10):
    t = threading.Thread(target=func)
    t.start()
    t_list.append(t)

for t in t_list:
    t.join()

print(‘end‘)

通过锁机制，同一时间只允许一个线程进行值的修改

import time
import threading

NUM = 10

def func(l):
    global NUM
    # 加锁
    l.acquire()
    NUM -= 1
    time.sleep(2)
    print(NUM)
    # 释放锁
    l.release()

lock = threading.Lock()

for i in range(10):
    t = threading.Thread(target=func, args=(lock,))
    t.start()

Semaphore(信号量)

锁允许一个线程在同一时间更改数据，而Semaphore是同时允许一定数量的线程更改数据。

import time
import threading

NUM = 10

def func(s):
    global NUM
    # 加锁
    s.acquire()
    NUM -= 1
    time.sleep(2)
    print(NUM)
    # 释放锁
    s.release()

lock = threading.Lock()
semaphore = threading.BoundedSemaphore(3)

for i in range(10):
    t = threading.Thread(target=func, args=(semaphore,))
    t.start()

事件（event）

python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。

event = threading.Event()

event.wait()

event.set()

event.clear()

import threading
import time

event = threading.Event()

def ligher():
    count = 0
    while True:
        if count < 30:
            if not event.is_set():
                event.set()
            print(‘green‘)
        elif count < 35:
            print(‘yellow‘)
        elif count < 60:
            if event.is_set():
                event.clear()
            print(‘red‘)
        else:
            count = 0
        count += 1
        time.sleep(0.3)


def car(n):
    count = 0
    while True:
        event.wait()
        print("car [%s] is running.." % n)
        count += 1
        time.sleep(1)

red_light = threading.Thread(target=ligher)
red_light.start()

c1 = threading.Thread(target=car, args=(1,))
c1.start()

 

条件（Condition）

 

Timer

from threading import Timer

def hello():
    print("hello, world")

t = Timer(1, hello)
t.start()  # after 1 seconds, "hello, world" will be printed

生产者消费者模型

例1:

import queue
import time
import threading


def consumer(name):
    while True:
        print("%s-->取到骨头[%s]" % (name, q.get()))
        time.sleep(0.5)


def producer(name):
    count = 0
    while q.qsize() < 5:
        print("%s 生产了骨头" % name, count)
        q.put(count)
        count += 1
        time.sleep(3)


q = queue.Queue(4)

p1 = threading.Thread(target=producer, args=(‘生产者1‘,))
p2 = threading.Thread(target=producer, args=(‘生产者2‘,))
c1 = threading.Thread(target=consumer, args=(‘消费者1‘,))
p1.start()
p2.start()
c1.start()

 例2:

import queue
import time
import threading


def consumer(name):
    while True:
        print("%s-->取到骨头[%s]" % (name, q.get()))
        time.sleep(0.5)
        q.task_done()  # 给生产者发一个回执，这个参数跟q.join联合是用


def producer(name):
    count = 0
    for i in range(10):
        print("%s 生产了骨头" % name, count)
        q.put(count)
        count += 1
        time.sleep(0.3)
    q.join()
    print(‘-----------所有骨头都吃完了-----‘)

q = queue.Queue(4)

p1 = threading.Thread(target=producer, args=(‘生产者1‘,))
# p2 = threading.Thread(target=producer, args=(‘生产者2‘,))
c1 = threading.Thread(target=consumer, args=(‘消费者1‘,))
p1.start()
# p2.start()
c1.start()

多进程multiprocessing

1、开销大

2、可以利用cpu的多核特性

 :

进程间的通信

不同进程间内存是不共享的，要想实现两个进程间的数据交换，可以用以下方法:

只是实现了数据的传递

Queue:

from multiprocessing import Process
from multiprocessing import Queue

def f(q):
    q.put([42, None, ‘hello‘])

if __name__ == ‘__main__‘:
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())  # prints "[42, None, ‘hello‘]"
    p.join()

Pipes

 

Managers

2个进程修改同一个数据

from multiprocessing import Process, Manager


def f(d, l,n):
    d[n] = n
    l.append(n)
    print(l)


if __name__ == ‘__main__‘:
    with Manager() as manager:
        d = manager.dict()

        l = manager.list(range(5))
        p_list = []
        for i in range(10):
            p = Process(target=f, args=(d, l, i))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()

        print(d)
        print(l)

进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池中有两个方法：

• apply
• apply_async

apply:串行基本没用callback方法

from multiprocessing import Pool
import time

def f(i):
    print(‘hello world‘, i)
    time.sleep(1)


def callback(data):
    print("exec done--->", data)


if __name__ == ‘__main__‘:

    pool = Pool(5)
    for num in range(10):
        pool.apply(func=f, args=(num,))

apply_async:

from multiprocessing import Pool
import time
import os


def f(i):
    print(‘hello world‘, i)
    time.sleep(1)
    print(‘-->PID‘, i, os.getpid())
    return i


def callback(data):  # 接收f()函数的返回值
    print(‘-->callback > pid‘, data, os.getpid())
    # print("exec done--->", data)


if __name__ == ‘__main__‘:

    pool = Pool(5)
    for num in range(100):
        pool.apply_async(func=f, args=(num,), callback=callback)

    # 必须有下边的  pool.close()和 pool.join()，别问为啥
    print(‘end‘)
    pool.close()
    pool.join()  # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。