Python实现线程池之线程安全队列

目录
  • 一、线程池组成
  • 二、线程安全队列的实现
  • 三、测试逻辑
    • 3.1、测试阻塞逻辑
    • 3.2、测试读写加锁逻辑

本文实例为大家分享了Python实现线程池之线程安全队列的具体代码,供大家参考,具体内容如下

一、线程池组成

一个完整的线程池由下面几部分组成,线程安全队列、任务对象、线程处理对象、线程池对象。其中一个线程安全的队列是实现线程池和任务队列的基础,本节我们通过threading包中的互斥量threading.Lock()和条件变量threading.Condition()来实现一个简单的、读取安全的线程队列。

二、线程安全队列的实现

包括put、pop、get等方法,为保证线程安全,读写操作时要添加互斥锁;并且pop操作可以设置等待时间以阻塞当前获取元素的线程,当新元素写入队列时通过条件变量通知解除等待操作。

class ThreadSafeQueue(object):

    def __init__(self, max_size=0):
        self.queue = []
        self.max_size = max_size  # max_size为0表示无限大
        self.lock = threading.Lock()  # 互斥量
        self.condition = threading.Condition()  # 条件变量

    def size(self):
        """
        获取当前队列的大小
        :return: 队列长度
        """
        # 加锁
        self.lock.acquire()
        size = len(self.queue)
        self.lock.release()
        return size

    def put(self, item):
        """
        将单个元素放入队列
        :param item:
        :return:
        """
        # 队列已满 max_size为0表示无限大
        if self.max_size != 0 and self.size() >= self.max_size:
            return ThreadSafeException()

        # 加锁
        self.lock.acquire()
        self.queue.append(item)
        self.lock.release()
        self.condition.acquire()
        # 通知等待读取的线程
        self.condition.notify()
        self.condition.release()

        return item

    def batch_put(self, item_list):
        """
        批量添加元素
        :param item_list:
        :return:
        """
        if not isinstance(item_list, list):
            item_list = list(item_list)

        res = [self.put(item) for item in item_list]

        return res

    def pop(self, block=False, timeout=0):
        """
        从队列头部取出元素
        :param block: 是否阻塞线程
        :param timeout: 等待时间
        :return:
        """
        if self.size() == 0:
            if block:
                self.condition.acquire()
                self.condition.wait(timeout)
                self.condition.release()
            else:
                return None

        # 加锁
        self.lock.acquire()
        item = None
        if len(self.queue):
            item = self.queue.pop()
        self.lock.release()

        return item

    def get(self, index):
        """
        获取指定位置的元素
        :param index:
        :return:
        """
        if self.size() == 0 or index >= self.size():
            return None

        # 加锁
        self.lock.acquire()
        item = self.queue[index]
        self.lock.release()

        return item

class ThreadSafeException(Exception):
    pass

三、测试逻辑

3.1、测试阻塞逻辑

def thread_queue_test_1():
    thread_queue = ThreadSafeQueue(10)

    def producer():
        while True:
            thread_queue.put(random.randint(0, 10))
            time.sleep(2)

    def consumer():
        while True:
            print('current time before pop is %d' % time.time())
            item = thread_queue.pop(block=True, timeout=3)
            # item = thread_queue.get(2)
            if item is not None:
                print('get value from queue is %s' % item)
            else:
                print(item)
            print('current time after pop is %d' % time.time())

    t1 = threading.Thread(target=producer)
    t2 = threading.Thread(target=consumer)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

测试结果:

我们可以看到生产者线程每隔2s向队列写入一个元素,消费者线程当无数据时默认阻塞3s。通过执行时间发现消费者线程确实发生了阻塞,当生产者写入数据时结束当前等待操作。

3.2、测试读写加锁逻辑

def thread_queue_test_2():
    thread_queue = ThreadSafeQueue(10)

    def producer():
        while True:
            thread_queue.put(random.randint(0, 10))
            time.sleep(2)

    def consumer(name):
        while True:
            item = thread_queue.pop(block=True, timeout=1)
            # item = thread_queue.get(2)
            if item is not None:
                print('%s get value from queue is %s' % (name, item))
            else:
                print('%s get value from queue is None' % name)

    t1 = threading.Thread(target=producer)
    t2 = threading.Thread(target=consumer, args=('thread1',))
    t3 = threading.Thread(target=consumer, args=('thread2',))
    t1.start()
    t2.start()
    t3.start()
    t1.join()
    t2.join()
    t3.join()

测试结果:

生产者还是每2s生成一个元素写入队列,消费者开启两个线程进行消费,默认阻塞时间为1s,打印结果显示通过加锁确保每次只有一个线程能获取数据,保证了线程读写的安全。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • Python线程之线程安全的队列Queue

    目录 一.什么是队列? 二.队列基操入队/出队/查队列状态 三.Queue是一个线程安全的类 一.什么是队列? 像排队一样,从头到尾排成一排,还可以有人继续往后排队,这就是队列. 这里学委想说的是Queue这个类, 它是queue这个内置模块内的一个类. import queue q = queue.Queue(5) #可以传入参数指定队列大小 queue.Queue()# 不传或者给0或者<0的数字则创建一个无限长度的队列 它提供了很多函数,下面几个函数,我们使用的比较多: get: 获取并移

  • Python线程之认识线程安全 

    目录 一.什么是线程安全? 二.在Python中有哪些类是线程安全的? 三.如何做到真正线程安全? 1.无状态函数 2.另一种化繁为简 一.什么是线程安全? 线程安全,名字就非常直接,在多线程情况下是安全的,多线程操作上的安全. 比如一个计算加法的函数,不管是一千个还是一万个线程,我们希望它执行的结果总是正确的,1+1 必须永远等于2, 而不是线程少的时候1+1 变成3或者4了. 通常我们都用线程安全来修饰一个类,修饰一个函数: 我们会说我设计的这个类是线程安全的这意味着,在多线程环境下,同时调

  • 线程安全及Python中的GIL原理分析

    本文讲述了线程安全及Python中的GIL.分享给大家供大家参考,具体如下: 摘要 什么是线程安全? 为什么python会使用GIL的机制? 在多核时代的到来的背景下,基于多线程来充分利用硬件的编程方法也不断发展起来, 但是一旦 牵扯到多线程,就必然会涉及到一个概念,即 线程安全, 本文就主要谈下笔者对线程安全的一些理解. 而Python为很多人所抱怨的一点就是GIL,那么python为什么选择使用GIL, 本文也就这个问题进行一些讨论. 引入 你的PC或者笔记本还是单核吗? 如果是,那你已经o

  • python多线程的线程如何安全实现

    1.引言 当前随着计算机硬件的快速发展,个人电脑上的 CPU 也是多核的,现在普遍的 CUP 核数都是 4 核或者 8 核的.因此,在编写程序时,需要为了提高效率,充分发挥硬件的能力,则需要编写并行的程序.Java 语言作为互联网应用的主要语言,广泛应用于企业应用程序的开发中,它也是支持多线程(Multithreading)的,但多线程虽好,却对程序的编写有较高的要求. 单线程可以正确运行的程序不代表在多线程场景下能够正确运行,这里的正确性往往不容易被发现,它会在并发数达到一定量的时候才可能出现

  • 详解python实现线程安全的单例模式

    单例模式是一种常见的设计模式,该模式的主要目的是确保某一个类只有一个实例存在.当你希望在整个系统中,某个类只能出现一个实例时,单例对象就能派上用场. 比如,服务器的配置信息写在一个文件中online.conf中,客户端通过一个 Config 的类来读取配置文件的内容.如果在程序运行期间,有很多地方都需要使用配置文件的内容,那么每个调用配置文件的地方都会创建 Config的实例,这就导致系统中存在多个Config 的实例对象,在配置文件内容很多的情况下,我们就浪费了大量的内存做了同样的事.事实上,

  • python线程安全及多进程多线程实现方法详解

    进程和线程的区别 进程是对运行时程序的封装,是系统资源调度和分配的基本单位 线程是进程的子任务,cpu调度和分配的基本单位,实现进程内并发. 一个进程可以包含多个线程,线程依赖进程存在,并共享进程内存 什么是线程安全 一个线程的修改被另一个线程的修改覆盖掉. python中哪些操作是线程安全的 一个操作可以在多线程环境中使用,并且获得正确的结果. 线程安全的操作线程是顺序执行的而不是并发执行的. 一般涉及到写操作需要考虑如何让多个线程安全访问数据. 线程同步的方式 互斥量(锁): 通过互斥机制防

  • Python实现线程池之线程安全队列

    目录 一.线程池组成 二.线程安全队列的实现 三.测试逻辑 3.1.测试阻塞逻辑 3.2.测试读写加锁逻辑 本文实例为大家分享了Python实现线程池之线程安全队列的具体代码,供大家参考,具体内容如下 一.线程池组成 一个完整的线程池由下面几部分组成,线程安全队列.任务对象.线程处理对象.线程池对象.其中一个线程安全的队列是实现线程池和任务队列的基础,本节我们通过threading包中的互斥量threading.Lock()和条件变量threading.Condition()来实现一个简单的.读

  • python自定义线程池控制线程数量的示例

    1.自定义线程池 import threading import Queue import time queue = Queue.Queue() def put_data_in_queue(): for i in xrange(10): queue.put(i) class MyThread(threading.Thread): def run(self): while not queue.empty(): sleep_times = queue.get() time.sleep(sleep_t

  • Java concurrency线程池之线程池原理(一)_动力节点Java学院整理

    ThreadPoolExecutor简介 ThreadPoolExecutor是线程池类.对于线程池,可以通俗的将它理解为"存放一定数量线程的一个线程集合.线程池允许若个线程同时允许,允许同时运行的线程数量就是线程池的容量:当添加的到线程池中的线程超过它的容量时,会有一部分线程阻塞等待.线程池会通过相应的调度策略和拒绝策略,对添加到线程池中的线程进行管理." ThreadPoolExecutor数据结构 ThreadPoolExecutor的数据结构如下图所示: 各个数据在Thread

  • Java concurrency线程池之线程池原理(二)_动力节点Java学院整理

    线程池示例 在分析线程池之前,先看一个简单的线程池示例. import java.util.concurrent.Executors; import java.util.concurrent.ExecutorService; public class ThreadPoolDemo1 { public static void main(String[] args) { // 创建一个可重用固定线程数的线程池 ExecutorService pool = Executors.newFixedThre

  • Java concurrency线程池之线程池原理(四)_动力节点Java学院整理

    拒绝策略介绍 线程池的拒绝策略,是指当任务添加到线程池中被拒绝,而采取的处理措施. 当任务添加到线程池中之所以被拒绝,可能是由于:第一,线程池异常关闭.第二,任务数量超过线程池的最大限制. 线程池共包括4种拒绝策略,它们分别是:AbortPolicy, CallerRunsPolicy, DiscardOldestPolicy和DiscardPolicy. AbortPolicy         -- 当任务添加到线程池中被拒绝时,它将抛出 RejectedExecutionException

  • Java 自定义线程池和线程总数控制操作

    1 概述 池化是常见的思想,线程池是非常典型的池化的实现,<Java并发编程实战>也大篇幅去讲解了Java中的线程池.本文实现一个简单的线程池. 2 核心类 [1]接口定义 public interface IThreadPool<Job extends Runnable> { /** * 关闭线程池 */ public void shutAlldown(); /** * 执行任务 * * @param job 任务 */ public void execute(Job job);

  • @Async异步线程池以及线程的命名方式

    本文记录@Async的基本使用以及通过实现ThreadFactory来实现对线程的命名. @Async的基本使用 近日有一个道友提出到一个问题,大意如下: 业务场景需要进行批量更新,已有数据id主键.更新的状态.单条更新性能太慢,所以使用in进行批量更新.但是会导致锁表使得其他业务无法访问该表,in的量级太低又导致性能太慢. 龙道友提出了一个解决方案,把要处理的数据分成几个list之后使用多线程进行数据更新.提到多线程可直接使用@Async注解来进行异步操作. 好的,接下来上面的问题我们不予解答

  • Java线程池复用线程的秘密你知道吗

    目录 前言 源码探究 execute方法 addWorker方法 Worker类 实现了Runnable接口 重要属性 构造方法 run方法 执行流程 总结 前言 我们都知道线程池可以帮我们管理线程,重复利用线程执行不同的任务.正常情况下,我们创建的线程执行完任务后就会自行销毁,那么线程池是如何做到复用线程的呢? 源码探究 我们从线程池ThreadPoolExecutor源码入手,一探究竟.为了突出重点,以下的方法源码过滤了部分无关代码,以求逻辑清晰. execute方法 那就从线程池执行的ex

  • java线程池中线程数量到底是几

    目录 线程池配置 线程池里的业务线程数量小于最小数量(5) 第一个请求 第二个请求 第三个请求 第五个请求 小于阻塞队列容量(10) 第六个请求 第七个请求 第15个请求 小于最大数量(20) 第16个请求 第35个请求 拒绝策略 第36个请求 复用线程 线程池配置 线程池配置,假设是: 1.最小数量是5 2.阻塞队列容量是10 3.最大数量是20 线程池里的业务线程数量小于最小数量(5) 第一个请求 第一个请求进来的时候,这个时候,线程池没有线程,就创建新的工作线程(即Worker线程). 然

  • Java concurrency线程池之线程池原理(三)_动力节点Java学院整理

    线程有5种状态:新建状态,就绪状态,运行状态,阻塞状态,死亡状态.线程池也有5种状态:然而,线程池不同于线程,线程池的5种状态是:Running, SHUTDOWN, STOP, TIDYING, TERMINATED. 线程池状态定义代码如下: private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3

随机推荐

其他