python小demo01:线程池+多进程实现cpu密集型操作

# -*- coding: utf8 -*-
import os
import sys
import time
import signal
import traceback# from shadowsocks
def write_pid_file(pid_file, pid):import fcntlimport stattry:fd &＃61; os.open(pid_file, os.O_RDWR | os.O_CREAT,stat.S_IRUSR | stat.S_IWUSR)except OSError:traceback.print_exc()return -1flags &＃61; fcntl.fcntl(fd, fcntl.F_GETFD)assert flags !&＃61; -1flags |&＃61; fcntl.FD_CLOEXECr &＃61; fcntl.fcntl(fd, fcntl.F_SETFD, flags)assert r !&＃61; -1# There is no platform independent way to implement fcntl(fd, F_SETLK, &fl)# via fcntl.fcntl. So use lockf insteadtry:fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB, 0, 0, os.SEEK_SET)except IOError:r &＃61; os.read(fd, 32)if r:print(&＃39;already started at pid %s&＃39; % (r))else:print(&＃39;already started&＃39;)os.close(fd)return -1os.ftruncate(fd, 0)os.write(fd, (str(pid)))return 0def freopen(f, mode, stream):oldf &＃61; open(f, mode)oldfd &＃61; oldf.fileno()newfd &＃61; stream.fileno()os.close(newfd)os.dup2(oldfd, newfd)def daemon_start(settings, main_process_handler):def handle_exit(signum, _):if signum &＃61;&＃61; signal.SIGTERM:sys.exit(0)sys.exit(1)signal.signal(signal.SIGINT, handle_exit)signal.signal(signal.SIGTERM, handle_exit)pid &＃61; os.fork()assert pid !&＃61; -1# Parentif pid:time.sleep(3)sys.exit(0)print("child has forked")# child signals its parent to exitppid &＃61; os.getppid()pid &＃61; os.getpid()if write_pid_file(settings.PID_FILE, pid) !&＃61; 0:os.kill(ppid, signal.SIGINT)sys.exit(1)# set self to process-group-leader
os.setsid()signal.signal(signal.SIGHUP, signal.SIG_IGN)print(&＃39;started&＃39;)os.kill(ppid, signal.SIGTERM)# octal 022os.umask(18)sys.stdin.close()try:freopen(settings.DEBUG_LOG_PATH, &＃39;a&＃39;, sys.stdout)freopen(settings.DEBUG_LOG_PATH, &＃39;a&＃39;, sys.stderr)except IOError:print(traceback.print_exc())sys.exit(1)main_process_handler()def daemon_stop(pid_file):import errnotry:with open(pid_file) as f:pid &＃61; buf &＃61; f.read()if not buf:print(&＃39;not running&＃39;)except IOError as e:print(traceback.print_exc())if e.errno &＃61;&＃61; errno.ENOENT:print("not running")# always exit 0 if we are sure daemon is not runningreturnsys.exit(1)pid &＃61; int(pid)if pid > 0:try:os.kill(pid, signal.SIGTERM)except OSError as e:if e.errno &＃61;&＃61; errno.ESRCH:print(&＃39;not running&＃39;)# always exit 0 if we are sure daemon is not runningreturnprint(traceback.print_exc())sys.exit(1)else:print(&＃39;pid is not positive: %d&＃39;, pid)# sleep for maximum 300sfor i in range(0, 100):try:# query for the pid
os.kill(pid, 0)except OSError as e:# not found the processif e.errno &＃61;&＃61; errno.ESRCH:breaktime.sleep(3)print("waiting for all threads finished....")else:print(&＃39;timed out when stopping pid %d&＃39;, pid)sys.exit(1)print(&＃39;stopped&＃39;)os.unlink(pid_file)def main():args &＃61; sys.argv[1:]assert len(args) &＃61;&＃61; 2if args[0] not in ["stop", "start"]:print("only supported: [stop | start]")returnif args[1] not in ["dev", "local", "prod"]:print("only supported: [dev | local | prod]")from globals import set_settings, initialize_redisset_settings(args[1])initialize_redis()from globals import settingsimport entryif args[0] &＃61;&＃61; "stop":print("stopping...")daemon_stop(settings.PID_FILE)elif args[0] &＃61;&＃61; "start":print("starting...")daemon_start(settings, entry.run)main()

# -*- coding: utf8 -*-
import json
import traceback
import signal
import sys
import time
from threading import Lock
from concurrent.futures import ThreadPoolExecutor
from .globals import settings, video_info_queuedef handler(data):# 业务逻辑
running_futures_count &＃61; 0def run():global running_futures_countcount_lock &＃61; Lock()pool &＃61; ThreadPoolExecutor(max_workers&＃61;settings.MAX_WORKER)try:def reduce_count(_):global running_futures_countwith count_lock:running_futures_count -&＃61; 1def handle_exit(_, __):print("get SIGINT signal")pool.shutdown(False)while True:if running_futures_count &＃61;&＃61; 0:sys.exit(0)time.sleep(1)print("now running futures count is %s, please wait" % running_futures_count)def handle_data(data):global running_futures_countwith count_lock:running_futures_count &＃43;&＃61; 1future &＃61; pool.submit(handler, data)future.add_done_callback(reduce_count)signal.signal(signal.SIGINT, handle_exit)signal.signal(signal.SIGTERM, handle_exit)while not pool._shutdown:print(len(pool._work_queue.queue), pool._shutdown)while not pool._shutdown and (len(pool._work_queue.queue) < pool._max_workers):data &＃61; video_info_queue.bpop(20)if data:handle_data(data)else:data &＃61; abnormal_video_info_queue.bpop(1)print("video_info_queue is empty, get data: %s from abnormal_video_info_queue" % data)if data:print("abnormal_video_info_queue")handle_data(data)time.sleep(5)print("now all the workers is busy, so wait and do not submit")finally:pool.shutdown(False)