python线程安全的计数器_python多线程的原理和实现

申明下哈 本篇文章不是自己写的 根据网上的文章再加上自己的加加点点 反正大部分都是网站的智慧哈&＃xff01;&＃xff01;&＃xff01;

1、线程基本概念

1.1 线程是什么&＃xff1f;

线程是指进程内的一个执行单元,也是进程内的可调度实体.

与进程的区别:

(1) 地址空间:进程内的一个执行单元;进程至少有一个线程;它们共享进程的地址空间;而进程有自己独立的地址空间;

(2) 资源拥有:进程是资源分配和拥有的单位,同一个进程内的线程共享进程的资源

(3) 线程是处理器调度的基本单位,但进程不是.

(4) 二者均可并发执行.

简而言之,一个程序至少有一个进程,一个进程至少有一个线程.

线程的划分尺度小于进程&＃xff0c;使得多线程程序的并发性高。

另外&＃xff0c;进程在执行过程中拥有独立的内存单元&＃xff0c;而多个线程共享内存&＃xff0c;从而极大地提高了程序的运行效率。

1.2 线程和进程关系&＃xff1f;

​ 进程就是一个应用程序在处理机上的一次执行过程&＃xff0c;它是一个动态的概念&＃xff0c;而线程是进程中的一部分&＃xff0c;进程包含多个线程在运行。

​ 多线程可以共享全局变量&＃xff0c;多进程不能。多线程中&＃xff0c;所有子线程的进程号相同&＃xff1b;多进程中&＃xff0c;不同的子进程进程号不同。

​ 进程是具有一定独立功能的程序关于某个数据集合上的一次运行活动,进程是系统进行资源分配和调度的一个独立单位.

​ 线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈),但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源.

​ 一个线程可以创建和撤销另一个线程;同一个进程中的多个线程之间可以并发执行.

2、Python线程模块

​ python主要是通过thread和threading这两个模块来实现多线程支持。python的thread模块是比较底层的模块&＃xff0c;python的threading模块是对thread做了一些封装&＃xff0c;可以更加方便的被使用。但是python(cpython)由于GIL的存在无法使用threading充分利用CPU资源&＃xff0c;如果想充分发挥多核CPU的计算能力需要使用multiprocessing模块(Windows下使用会有诸多问题)。

2.1 如何创建线程

我自己不懂python2 所以直接拿python3来写。。。

python3.x中通过threading模块创建新的线程有两种方法&＃xff1a;

(1)第一种是通过threading.Thread(Target&＃61;executable Method)-即传递给Thread对象一个可执行方法(或对象)

(2)第二种是继承threading.Thread定义子类并重写run()方法。第二种方法中&＃xff0c;唯一必须重写的方法是run()

实例&＃xff1a;通过threading.Thread进行创建多线程

import threading

import time

def target():

print("当前线程名为&＃xff1a;%s 正在运行中"

% (threading.current_thread().name))

time.sleep(1)

print("当前线程名为&＃xff1a;%s 正在运行中" %

(threading.current_thread().name))

print("当前线程名为&＃xff1a;%s 正在运行中" % (threading.current_thread().name))

# 属于线程t的部分

t &＃61; threading.Thread(target&＃61;target)

t.start()

# 属于线程t的部分

t.join() # join是阻塞当前线程(此处的当前线程时主线程) 主线程直到Thread-1结束之后才结束

print("当前线程名为&＃xff1a;%s 已经结束" % (threading.current_thread().name))

此时运行的线程一共有两个&＃xff0c;一个主线程一个子线程!!!

其实我们在这里还可以探讨下为什么最后是MianThread最后结束的呢&＃xff1f;

答&＃xff1a;原因是因为有join函数

join函数的作用是&＃xff1a;当遇到join的时候使主线程阻塞&＃xff0c;等子线程运行完成后再继续主线程&＃xff0c;这样我们就可以获得两个线程并行运行的时间了。只需要对子线程调用join方法就可以将主线程阻塞

那么我们有没有办法使得主线程退出后子线程自动退出呢(主线程不需要等子线程完成后再继续往下走)&＃xff1f;

答&＃xff1a;通过对子线程调用setDaemon方法&＃xff0c;将子线程标记成守护线程

实例&＃xff1a;

# coding&＃61;utf-8

import threading

import time

def to_f(time_time_time):

time.sleep(2)

print(&＃39;子线程%s完成了哈&＃xff01;&＃xff01;完成时间->%s&＃39; %

(threading.current_thread().name, time_time_time))

for i in range(2):

t &＃61; threading.Thread(target&＃61;to_f, args&＃61;(time.time(),))

t.setDaemon(1)

t.start()

print(&＃39;主线程完成&＃xff01;&＃xff01;&＃xff01;&＃39;)

这里的话不需要等待子线程了 直接完成 由于主线程运行到结尾的时候 子线程还在sleep 所以直接结束子线程&＃xff01;&＃xff01;&＃xff01;

再看这个&＃xff1a;通过继承threading.Thread定义子类创建多线程

使用Threading模块创建线程&＃xff0c;直接从threading.Thread继承&＃xff0c;然后重写init方法和run方法&＃xff1a;

import threading

import time

class myThread(threading.Thread): # 继承父类threading.Thread

def __init__(self, threadID, name, counter):

threading.Thread.__init__(self)

self.threadID &＃61; threadID

self.name &＃61; name

self.counter &＃61; counter

def run(self): # 把要执行的代码写到run函数里面 线程在创建后会直接运行run函数

print("Starting " &＃43; self.name)

print_time(self.name, self.counter, 5)

print("Exiting " &＃43; self.name)

def print_time(threadName, delay, counter):

while counter:

time.sleep(delay)

print("%s process at: %s" % (threadName, time.ctime(time.time())))

counter -&＃61; 1

# 创建新线程

thread1 &＃61; myThread(1, "Thread-1", 1)

thread2 &＃61; myThread(2, "Thread-2", 2)

# 开启线程

thread1.start()

thread2.start()

# 等待线程结束

thread1.join()

thread2.join()

print("Exiting Main Thread")

结果&＃xff1a;

结果发现子线程-1 子线程-2 不是有序的进行执行 而是看来像抢着执行的 但是由于thread-1 start()先开始的 所以thread-1优先结束的可能性大哈&＃xff01;&＃xff01;&＃xff01; 我们也知道多线程之间的变量是共享的 如果如上方的情况发生那数据岂不是会直接混乱起来 那么有啥方法呢&＃xff1f;

3、线程间同步

如果多个线程共同对某个数据修改&＃xff0c;则可能出现不可预料的结果&＃xff0c;为了保证数据的正确性&＃xff0c;需要对多个线程进行同步。

​ 使用Thread对象的Lock和Rlock可以实现简单的线程同步&＃xff0c;这两个对象都有acquire方法和release方法&＃xff0c;对于那些需要每次只允许一个线程操作的数据&＃xff0c;可以将其操作放到acquire和release方法之间。

​ 需要注意的是&＃xff0c;Python有一个GIL(Global Interpreter Lock)机制&＃xff0c;任何线程在运行之前必须获取这个全局锁才能执行&＃xff0c;每当执行完100条字节码&＃xff0c;全局锁才会释放&＃xff0c;切换到其他线程执行。

3.1 线程同步问题

多线程实现同步有四种方式&＃xff1a;

锁机制&＃xff0c;信号量&＃xff0c;条件判断和同步队列。

下面我主要关注两种同步机制&＃xff1a;锁机制和同步队列。

(1)锁机制

threading的Lock类&＃xff0c;用该类的acquire函数进行加锁&＃xff0c;用realease函数进行解锁

threadLock &＃61; threading.Lock() # threading的Lock类来获得一个锁

代码如下&＃xff1a;

import threading

import time

class myThread(threading.Thread): # 继承父类threading.Thread

def __init__(self, threadID, name, sleep_time):

threading.Thread.__init__(self)

self.threadID &＃61; threadID

self.name &＃61; name

self.sleep_time &＃61; sleep_time

def run(self): # 把要执行的代码写到run函数里面 线程在创建后会直接运行run函数

print("Starting " &＃43; self.name)

threadLock.acquire()

print_time(self.name, self.sleep_time, 5)

print("Exiting " &＃43; self.name)

threadLock.release()

def print_time(threadName, delay, counter):

while counter:

time.sleep(delay)

print("%s process at: %s" % (threadName, time.ctime(time.time())))

counter -&＃61; 1

threadLock &＃61; threading.Lock()

# 创建新线程

thread1 &＃61; myThread(1, "Thread-1", 1)

thread2 &＃61; myThread(2, "Thread-2", 2)

# 开启线程

thread1.start()

thread2.start()

# 等待线程结束

thread1.join()

thread2.join()

print("Exiting Main Thread")

这里对于锁的理解&＃xff1a;

假如有两个线程A和B&＃xff0c;A和B里的程序都加了同一个锁对象,当线程A率先执行到lock.acquire() (ps.拿到全局唯一的锁后).

线程B只能等到线程A释放锁lock.release()后(归还锁)才能运行lock.acquire()(拿到全局唯一的锁)并执行后面的代码

其中锁也可以这样实现

lock &＃61; threading.Lock()

with lock: # with语句会在这个代码块执行前自动获取锁&＃xff0c;在执行结束后自动释放锁

# 这里写自己的代码

pass

关于锁的内容还可以参考https://www.jianshu.com/p/9fa218e50a16

(2) 线程同步队列queue

用法&＃xff1a;import queue.

Python的queue模块中提供了同步的、线程安全的队列类&＃xff0c;包括FIFO(先入先出)队列Queue&＃xff0c;LIFO(后入先出)队列LifoQueue&＃xff0c;和优先级队列PriorityQueue。这些队列都实现了锁原语&＃xff0c;能够在多线程中直接使用。可以使用队列来实现线程间的同步。

queue模块中的常用方法:

queue.qsize() 返回队列的大小

queue.empty() 如果队列为空&＃xff0c;返回True,反之False

queue.full() 如果队列满了&＃xff0c;返回True,反之False

queue.full 与 maxsize 大小对应

queue.get([block[, timeout]])获取队列&＃xff0c;timeout等待时间

queue.get_nowait() 相当Queue.get(False)

queue.put(item) 写入队列&＃xff0c;timeout等待时间

queue.put_nowait(item) 相当Queue.put(item, False)

queue.task_done() 在完成一项工作之后&＃xff0c;Queue.task_done()函数向任务已经完成的队列发送一个信号

queue.join() 实际上意味着等到队列为空&＃xff0c;再执行别的操作

实例1

import queue

import threading

import time

exitFlag &＃61; 0

class myThread(threading.Thread):

def __init__(self, threadID, name, q):

threading.Thread.__init__(self)

self.threadID &＃61; threadID

self.name &＃61; name

self.q &＃61; q

def run(self):

print("Starting " &＃43; self.name)

process_data(self.name, self.q)

print("Exiting " &＃43; self.name)

def process_data(threadName, q):

while not exitFlag:

queueLock.acquire()

if not workQueue.empty():

data &＃61; q.get()

queueLock.release()

print("%s processing %s" % (threadName, data))

else:

queueLock.release()

time.sleep(1)

queueLock &＃61; threading.Lock()

workQueue &＃61; queue.Queue(100)

threads &＃61; []

threadID &＃61; 1

# 创建新线程

for tName in range(10): # 创建10个线程

thread &＃61; myThread(threadID, &＃39;线程-&＃39; &＃43; str(tName) &＃43; &＃39;号&＃39;, workQueue)

thread.start()

threads.append(thread)

threadID &＃43;&＃61; 1

# 填充队列

for word in range(1, 101): # 把100个任务放入队列中

workQueue.put(word)

# 等待队列清空

while not workQueue.empty():

pass

# 通知线程是时候退出

exitFlag &＃61; 1

# 等待所有线程完成

for t in threads:

t.join()

print("Exiting Main Thread")

实例2

import time

import threading

import queue

class Worker(threading.Thread):

def __init__(self, name, queue):

threading.Thread.__init__(self)

self.queue &＃61; queue

self.start() #执行run()

def run(self):

#循环&＃xff0c;保证接着跑下一个任务

while True:

# 队列为空则退出线程

if self.queue.empty():

break

# 获取一个队列数据

foo &＃61; self.queue.get()

# 延时1S模拟你要做的事情

time.sleep(1)

# 打印

print(self.getName() &＃43; " process " &＃43; str(foo))

# 任务完成

self.queue.task_done()

# 队列

queue &＃61; queue.Queue()

# 加入100个任务队列

for i in range(100):

queue.put(i)

# 开10个线程

for i in range(10):

threadName &＃61; &＃39;Thread&＃39; &＃43; str(i)

Worker(threadName, queue)

# 所有线程执行完毕后关闭

queue.join()