使用ThreadPoolExecutor and schedule进行创建线程池（多线程-定时任务-数据同步-任务队列）

使用ThreadPoolExecutor and schedule进行创建线程池（多线程-定时任务-数据同步-任务队列）

# 使用ThreadPoolExecutor and schedule进行创建线程池（多线程-定时任务-数据同步-任务队列）
from concurrent.futures import ThreadPoolExecutor#导入线程池库
import threading#导入线程库
import time#导入时间库
import schedule#导入调度库——定时
import random#随机数生成库引入
# --------------------------------------------------------------
# 定义一个任务函数（求和），用来多线程执行测试
def task(n):#这是一个作业函数，去执行任务的函数
    time.sleep(3)#让程序休眠3秒
    sum=0
    for i in range(n+1):#求和
        sum =sum+i
    # print(threading.current_thread().name,n,"的和为：",sum)#threading.current_thread().name可以获取到当前线程名称
    return str(n) + '的求和为：' + str(sum)#将执行结果返回
# --------------------------------------------------------------
# 创建一个函数任务（添加任务），用来向线程池的队列中【添加新任务】。
def addtask():#将task任务添加到线程池的队列中
    for i in range(5):#创建十个任务
        future1 = pool.submit(task, i)#将task任务【添加至线程池的队列中】
        threadlist.append(future1)#将线程对象放到列表中，便于【二次调用】
    for th in threadlist:#遍历线程对象
        print(th.result())#调用任务线程，获取任务执行完毕返回的【结果】
    # 清空列表，会变成[]
    threadlist.clear()
# --------------------------------------------------------------
# 创建一个函数任务（死循环-调用添加任务函数）：子线程执行该函数，进入死循环中，间断执行添加任务函数。
def queue01():
    while(True):#设立死循环，不让线程死亡，间断执行方法体的程序。
        addtask()#调用添加任务函数
        print('休眠3秒钟~')#向控制台打印日志
        try:
            time.sleep(3)#让程序休眠3秒
        except:
            pass#异常不执行，直接跳过（相当于空语句）
# --------------------------------------------------------------
#创建一个函数任务，用来子线程测试用，【仅有测试用途，无涉及其他内容调用】
def test(n):
    print(threading.current_thread().name,n)
# @@@【分隔符】-------------------------------------------------------------------------------
# 以上函数的实现与下列的函数功能相同，可以只看下面（或者看上面）的函数即可实现功能。建议看下面的较简单好理解。
# @@@---------------------------------------------------------------------------------------

# 线程同步~
def addvalue():
    lock.acquire()# 操作共享数据前，申请获取锁
    threadlist.append(int(random.random()*100))
    lock.release()# 操作完共享数据后，申请释放锁
# --------------------------------------------------------------
def job01():#定时执行-作业1
    print("I'm working...job01",threading.current_thread().name)
    addvalue()
    print(threadlist)
    time.sleep(1)
# --------------------------------------------------------------
def job02():#定时执行-作业2
    print("I'm working...job02",threading.current_thread().name)
    addvalue()
    print(threadlist)
    time.sleep(1)
# --------------------------------------------------------------
def job03():#定时执行-作业3
    print("I'm working...job03",threading.current_thread().name)
    addvalue()
    print(threadlist)
    time.sleep(1)
# --------------------------------------------------------------
def addtask01():#将作业1添加到队列中
    pool.submit(job01,)  # 将job01任务【添加至线程池的队列中】
# --------------------------------------------------------------
def addtask02():#将作业2添加到队列中
    pool.submit(job02,)  # 将job02任务【添加至线程池的队列中】
# --------------------------------------------------------------
def addtask03():#将作业3添加到队列中
    pool.submit(job03,)  # 将job03任务【添加至线程池的队列中】
# --------------------------------------------------------------

if __name__=='__main__':
    startime=time.time()#设置一个时间起点，用于计算程序执行【耗时长】
    threadlist=[]#声明一个列表，用于接受子线程
    lock=threading.Lock()#申明一把锁，用于多线程同步（多个线程访问同一个对象，一次只能一个线程访问该对象）
    # 创建一个线程池
    # pool = tp.ThreadPool(5)
    pool = ThreadPoolExecutor(max_workers = 2)  #定义线程池
    # 创建子线程
    # thread=threading.Thread(target=test,args=('one',))
    # thread=threading.Thread(target=queue01)#子线程执行queue01任务函数
    # thread.start()#启动子线程
    # 等待子线程结束，主线程才往下执行
    # thread.join()
    # addtask()#主线程：执行任务函数，往线程池中添加任务
    print(time.time()-startime)#打印程序执行消耗的时长
    schedule.every(1).seconds.do(addtask01)
    schedule.every(1).seconds.do(addtask02)
    schedule.every(1).seconds.do(addtask03)
    # schedule.every().day.at("16:35:00").do(addtask01)
    # schedule.every().day.at("16:35:02").do(addtask01)
    # schedule.every().day.at("16:35:03").do(addtask01)
    #
    # schedule.every().day.at("16:35:04").do(addtask02)
    # schedule.every().day.at("16:35:05").do(addtask02)
    # schedule.every().day.at("16:35:06").do(addtask02)
    #
    # schedule.every().day.at("16:35:07").do(addtask03)
    # schedule.every().day.at("16:35:08").do(addtask03)
    # schedule.every().day.at("16:35:09").do(addtask03)
    while True:
        time_s=random.random()*0.5+1
        print('--------------------------线程池队列排队任务数量：',pool._work_queue.qsize())#pool._work_queue.qsize()获取线程队列的任务数
        schedule.run_pending()
        try:
            time.sleep(time_s)
        except:
            pass