python 生产者消费者队列简单封装

返回上一步 / in 0

python 生产者消费者队列简单的封装

使用了python自带的threading和queue队列

# 封装线程类

class My_threadingpool:
    from queue import Queue
    def __init__(self):
        self.q=Queue()
    # 生产者
    def put_data_in_q(self):
        for x in range(10):
            self.q.put(x)
    # 需要执行的任务
    def task(self,data):
        print('执行者输出',data)
        time.sleep(1)
    # 消费者  消费者就是执行任务的人，方便理解把task封成了方法，看起来更清晰
    def get_data_in_q(self):
        while True:
            if self.q.qsize()==0:
                break
            else:
                data=self.q.get()
                # 执行任务，
                self.task(data)
    # 线程控制退出，避免出现还有余下的线程没有执行完就主线程就退出了
    def listening_quit(self):
        def listening():
            time.sleep(1)
            while 1:
                if len(threading.enumerate()) == 2:
                    break
                time.sleep(0.5)
        self.t=threading.Thread(target=listening,)
        self.t.start()
        # self.t.join() # 放在消费者线程后面
    # 其他

    # 线程开启,有了queue消息队列，不在使用线程传参
    def threadpool(self,num=3):
        for x in range(num):
            t=threading.Thread(target=self.get_data_in_q,)
            t.start()
        t.join()
        self.listening_quit()
        self.t.join()

# 例子，重写了父类的数据入队列的方法，同样可以重写task方法替换成你需要执行的任务过程
class Ji(My_threadingpool):
    def __init__(self):
        super(Ji, self).__init__()
    def put_data_in_q(self):
        for x in range(10,20):
            self.q.put(x)
    def task(self,data):
        print('改写task',data)
        time.sleep(2)
p=Ji()
p.put_data_in_q()
p.threadpool(5)

 

搜索