首页 > 代码库 > Python多线程、进程入门1

Python多线程、进程入门1

进程是资源的一个集合,

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()  # 进程池中进程执行完毕后再关闭,如果注释,那么程序直接关闭。

 

Python多线程、进程入门1