小白也能掌握的Python部署应用技术

from datetime import datetime
import time
from tools.sender_mail import Sender_mail
from tools.Wrapper_timer import time_logger
now_time = datetime.now().date()

#业务数据生产过程
def tick1():
    time.sleep(4)
    print('Tick!The time is:%s' % datetime.now())

#预警信息发送
def tick2():
    #发送邮件
    Sender_mail.sender_mail()
    print('发送完成')


# def get_pid(name):
#     '''
#      作用：根据进程名获取进程pid
#     '''
#     import psutil
#     pids = psutil.process_iter()
#     print("[" + name + "]'s pid is:")
#     for Pid in pids:
#         if(Pid.name() == name):
#             P=Pid.pid
#     return P

#关闭解释器
def end_program():
    # pr_name = get_pid(name="pythonw.exe")
    # os.system('%s%s' % ("taskkill /F /PID ",pr_name))
    #os.system('%s%s' % ("taskkill /F /IM pythonw.exe"))
    
    import subprocess
    CREATE_NO_WINDOW = 0x08000000
    subprocess.call('taskkill /F /IM pythonw.exe', creationflags=CREATE_NO_WINDOW)

#业务流程
@time_logger
def tick():
    tick1()
    tick2()
    end_program()

if __name__ =='__main__':
    tick()

def tick():
    print('job thread_id-{0}, process_id-{1}'.format(threading.get_ident(), os.getpid()))
    w = WorkerThread()
    w.start()
    w.send(tick1())
    w.send(tick2())
    w.run()
    w.send(end_program())
    w.run()
    w.close()

# 使用 队列一般可以简化多线程的程序。例如，可以使用共享队列将线程连接在一起，而不必依赖锁的保护。
# 在这种模型下，工作者线程一般充当数据的消费者。
from threading import Thread
from queue import Queue
import time,os,threading
class WorkerThread(Thread):
    def __init__(self,*args,**kwargs):
        Thread.__init__(self,*args,**kwargs)
        self.input_queue=Queue()

    def send(self,item):
        self.input_queue.put(item)

    def close(self):
        self.input_queue.put(None)
        self.input_queue.join()
    def run(self):
        while True:
            item=self.input_queue.get()
            if item is None:
                break
            #实际开发中，此处应该使用有用的工作代替
            print(item)
            self.input_queue.task_done()
        #完成，指示收到和返回哨兵
        self.input_queue.task_done()
        return

import functools
import logging
import time
from datetime import datetime

log_format = "%(asctime)s %(message)s"
logging.basicConfig(format=log_format,
                    level=logging.INFO,
                    datefmt=" %Y-%m-%d %H:%M:%S",  # 时间格式
                    filename="log1.log",
                    filemode="a")


# 装饰器会被任何函数使用。其中的func(*args, **kwargs)中的func就是目标函数，args、kwargs是这个函数调用的参数
def time_logger(func):
  @functools.wraps(func)
  def wrapper_timer(*args, **kwargs):
    logging.info(f"BeginFunction {func.__name__!r}, args={args}, kwargs={kwargs}")
    start_time = time.perf_counter()  # 1
    value = func(*args, **kwargs)
    end_time = time.perf_counter()  # 2
    run_time = end_time - start_time  # 3

    logging.info(f"EndFunction {func.__name__!r} in {run_time:.4f} secs")
    return value,run_time

  return wrapper_timer

from datetime import datetime
import time,os,threading
from tools.Wrapper_timer import time_logger
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor
from tools.Queue_thread import WorkerThread
import logging

now_time = datetime.now().date()
# 业务数据生产过程
@time_logger
def tick1():
    print('job1 thread_id-{0}, process_id-{1}'.format(threading.get_ident(), os.getpid()))
    time.sleep(1)
    print('Tick!The time is:%s' % datetime.now())


# 预警信息发送
@time_logger
def tick2():
    print('job2 thread_id-{0}, process_id-{1}'.format(threading.get_ident(), os.getpid()))
    # 发送邮件
    time.sleep(1)
    print('邮件发送完成 %s'% datetime.now())

#数据存储
@time_logger
def tick3():
    print('job3 thread_id-{0}, process_id-{1}'.format(threading.get_ident(), os.getpid()))
    # 数据存储
    time.sleep(1)
    print('存储完成 %s'% datetime.now())


def tick():
    print('job thread_id-{0}, process_id-{1}'.format(threading.get_ident(), os.getpid()))
    w = WorkerThread()
    w.start()
    w.send(tick1())
    w.send(tick2())
    w.run()
    w.send(tick3())
    w.run()
    w.close()

if __name__ == '__main__':
    executors = {
        'default': ThreadPoolExecutor(10)
    }
    scheduler = BlockingScheduler(executors=executors,)
    # scheduler.add_job(tick,max_instances=5,trigger='interval', seconds=10,id='interval_task')
    #注意装饰器@time_logger不能放在tick前面,否则会出现阻塞，程序无法继续运行
    scheduler.add_job(tick,max_instances=5,trigger='interval', seconds=10)
    # run_time=tick1()[1]
    # print("完成时间",run_time)
    # scheduler._logger=logging
    try:
        scheduler.start()
    except (KeyboardInterrupt,SystemExit):
        pass