商丘做网站的公司/网络整合营销方案ppt

一直苦于python及shell脚本的串并行总控调度问题；虽然可以通过crontab配置进行一些不太复杂的调度，但毕竟功能有限，也显得有点low；要么就是开源调度工具及收费的商用工具。
一般在项目不是很庞大，但也有一点复杂性方面，在这里考虑自己写代码，实现程序编码的调度。
首先，将要调度的可并行程序放在相同的目录下；调控程序根据目录下的脚本文件和设定的并行参数，进行并行调度；
然后，在调度的过程中有一些日志存储，便于问题的查找和发现；调度的时间设定参照crontab的格式，一样的方式配置；
再次，程序以后的优化：在图形化方面、串并行混合方面有很大的提升空间；这是一个初始版本，后期根据项目的需要进行相应必要的更新。
1、程序目录树

MacBook-Pro-Nisj:cronLake nisj$ find . -print | sed -e 's;[^/]*/;|____;g;s;____|; |;g'
.
|____scriptDir
| |____cron2
| | |____1.sh
| | |____2.py
| | |____1.py
| | |____2.sh
| |____cron1
| | |____2.py
| | |____1.py
|____threadpool.pyc
|____auxiliaryFunction.py
|____threadpool.py
|____parseCron.pyc
|____start.py
|____parseCron.py
|____cronConfig.py
|____log
| |____cronLake_run_2018-06-05.log.5
| |____cronLake_run_2018-06-05.log.2
| |____cronLake_run_2018-05-22.log
| |____cronLake_run_2018-06-05.log.3
| |____cronLake_run_2018-06-05.log.4
| |____cronLake_run_2018-06-05.log
| |____cronLake_run_2018-06-05.log.1
|____auxiliaryFunction.pyc
|____cronConfig.pyc
MacBook-Pro-Nisj:cronLake nisj$ pwd
/Users/nisj/PycharmProjects/BiDataProc/cronLake
MacBook-Pro-Nisj:cronLake nisj$ 

2、配置文件
/Users/nisj/PycharmProjects/BiDataProc/cronLake/cronConfig.py

# -*- coding=utf-8 -*-
cronDirMap = {"cron1": {"enable": "Ture","fileDir": "/Users/nisj/PycharmProjects/BiDataProc/cronLake/scriptDir/cron2","fileExt": "py","cron": "* * * * *","parallelNum": 3},"cron2": {"enable": "False","fileDir": "/Users/nisj/PycharmProjects/BiDataProc/cronLake/scriptDir/cron2","fileExt": "sh","cron": "* * * * *","parallelNum": 3}
}

其中：文件以json方式存取，涉及程序目录、程序类型、并行个数、cron时间配置及是否生效5个选项；根据后期需要可以再进行增减。
3、解析cron格式的脚本
/Users/nisj/PycharmProjects/BiDataProc/cronLake/parseCron.py

# /usr/bin/env python
# -*- coding:utf-8 -*-"""
1.解析 crontab 配置文件中的五个数间参数(分 时 日 月 周)，获取他们对应的取值范围
2.将时间戳与crontab配置中一行时间参数对比，判断该时间戳是否在配置设定的时间范围内
"""import re, time, sysdef match_cont(patten, cont):"""正则匹配(精确符合的匹配)Args:patten 正则表达式cont____ 匹配内容Return:True or False"""res = re.match(patten, cont)if res:return Trueelse:return Falsedef handle_num(val, ranges=(0, 100), res=list()):"""处理纯数字"""val = int(val)if val >= ranges[0] and val <= ranges[1]:res.append(val)return resdef handle_nlist(val, ranges=(0, 100), res=list()):"""处理数字列表 如 1,2,3,6"""val_list = val.split(',')for tmp_val in val_list:tmp_val = int(tmp_val)if tmp_val >= ranges[0] and tmp_val <= ranges[1]:res.append(tmp_val)return resdef handle_star(val, ranges=(0, 100), res=list()):"""处理星号"""if val == '*':tmp_val = ranges[0]while tmp_val <= ranges[1]:res.append(tmp_val)tmp_val = tmp_val + 1return resdef handle_starnum(val, ranges=(0, 100), res=list()):"""星号/数字 组合 如 */3"""tmp = val.split('/')val_step = int(tmp[1])if val_step < 1:return resval_tmp = int(tmp[1])while val_tmp <= ranges[1]:res.append(val_tmp)val_tmp = val_tmp + val_stepreturn resdef handle_range(val, ranges=(0, 100), res=list()):"""处理区间 如 8-20"""tmp = val.split('-')range1 = int(tmp[0])range2 = int(tmp[1])tmp_val = range1if range1 < 0:return reswhile tmp_val <= range2 and tmp_val <= ranges[1]:res.append(tmp_val)tmp_val = tmp_val + 1return resdef handle_rangedv(val, ranges=(0, 100), res=list()):"""处理区间/步长 组合 如 8-20/3 """tmp = val.split('/')range2 = tmp[0].split('-')val_start = int(range2[0])val_end = int(range2[1])val_step = int(tmp[1])if (val_step < 1) or (val_start < 0):return resval_tmp = val_startwhile val_tmp <= val_end and val_tmp <= ranges[1]:res.append(val_tmp)val_tmp = val_tmp + val_stepreturn resdef parse_conf(conf, ranges=(0, 100), res=list()):"""解析crontab 五个时间参数中的任意一个"""# 去除空格，再拆分conf = conf.strip(' ').strip(' ')conf_list = conf.split(',')other_conf = []number_conf = []for conf_val in conf_list:if match_cont(PATTEN['number'], conf_val):# 记录拆分后的纯数字参数number_conf.append(conf_val)else:# 记录拆分后纯数字以外的参数，如通配符 * , 区间 0-8, 及 0－8/3 之类other_conf.append(conf_val)if other_conf:# 处理纯数字外各种参数for conf_val in other_conf:for key, ptn in PATTEN.items():if match_cont(ptn, conf_val):res = PATTEN_HANDLER[key](val=conf_val, ranges=ranges, res=res)if number_conf:if len(number_conf) > 1 or other_conf:# 纯数字多于1，或纯数字与其它参数共存，则数字作为时间列表res = handle_nlist(val=','.join(number_conf), ranges=ranges, res=res)else:# 只有一个纯数字存在，则数字为时间 间隔res = handle_num(val=number_conf[0], ranges=ranges, res=res)return resdef parse_crontab_time(conf_string):"""解析crontab时间配置参数Args:conf_string  配置内容(共五个值：分 时 日 月 周)取值范围 分钟:0-59 小时:1-23 日期:1-31 月份:1-12 星期:0-6(0表示周日)Return:crontab_range    list格式，分 时 日 月 周 五个传入参数分别对应的取值范围"""time_limit = ((0, 59), (1, 23), (1, 31), (1, 12), (0, 6))crontab_range = []clist = []conf_length = 5tmp_list = conf_string.split(' ')for val in tmp_list:if len(clist) == conf_length:breakif val:clist.append(val)if len(clist) != conf_length:return -1, 'config error whith [%s]' % conf_stringcindex = 0for conf in clist:res_conf = []res_conf = parse_conf(conf, ranges=time_limit[cindex], res=res_conf)if not res_conf:return -1, 'config error whith [%s]' % conf_stringcrontab_range.append(res_conf)cindex = cindex + 1return 0, crontab_range# crontab时间参数各种写法 的 正则匹配
PATTEN = {# 纯数字'number': '^[0-9]+$',# 数字列表,如 1,2,3,6'num_list': '^[0-9]+([,][0-9]+)+$',# 星号 *'star': '^\*$',# 星号/数字 组合，如 */3'star_num': '^\*\/[0-9]+$',# 区间 如 8-20'range': '^[0-9]+[\-][0-9]+$',# 区间/步长 组合 如 8-20/3'range_div': '^[0-9]+[\-][0-9]+[\/][0-9]+$'# 区间/步长 列表 组合，如 8-20/3,21,22,34# 'range_div_list':'^([0-9]+[\-][0-9]+[\/][0-9]+)([,][0-9]+)+$'
}
# 各正则对应的处理方法
PATTEN_HANDLER = {'number': handle_num,'num_list': handle_nlist,'star': handle_star,'star_num': handle_starnum,'range': handle_range,'range_div': handle_rangedv
}def getCronList(conf_string):# 时间戳time_stamp = int(time.time())# 解析crontab时间配置参数 分 时 日 月 周 各个取值范围res, desc = parse_crontab_time(conf_string)if res == 0:cron_time = descelse:print descsys, exit(-1)# print "\nconfig:", conf_stringreturn {"minute": cron_time[0], "hour": cron_time[1], "day": cron_time[2], "month": cron_time[3],"week": cron_time[4]}# test
# if __name__ == '__main__':
#     cronDict = getCronList(conf_string='*/1 23 9-11 * 6')
#     print cronDict['week'], cronDict['month'], cronDict['day'], cronDict['hour'], cronDict['minute']

脚本从网上借鉴而来，进行了一些改动，方便实用。
4、附属功能脚本集合
/Users/nisj/PycharmProjects/BiDataProc/cronLake/auxiliaryFunction.py

# -*- coding=utf-8 -*-
import os
import datetime
import logging.handlersdef logRecord(scriptFile):currDate = datetime.datetime.now().strftime('%Y-%m-%d')logFile = 'log/cronLake_run_{currDate}.log'.format(currDate=currDate)handler = logging.handlers.RotatingFileHandler(logFile, maxBytes=1024 * 1024, backupCount=5)  # 实例化handlerfmt = '%(asctime)s - %(name)s - %(message)s-{scriptFile} '.format(scriptFile=scriptFile)formatter = logging.Formatter(fmt)  # 实例化formatterhandler.setFormatter(formatter)  # 为handler添加formatterlogger = logging.getLogger('cronLake_run')  # 获取名为cronLake_run的loggerlogger.addHandler(handler)  # 为logger添加handlerlogger.setLevel(logging.DEBUG)logger.info('{scriptFile} info message'.format(scriptFile=scriptFile))logger.debug('{scriptFile} debug message'.format(scriptFile=scriptFile))

在这里目前只有日志记录方面的处理。
5、主脚本
/Users/nisj/PycharmProjects/BiDataProc/cronLake/start.py

# -*- coding=utf-8 -*-
import os
import datetime
import threadpool
import time
from cronConfig import cronDirMap
from parseCron import getCronList
from auxiliaryFunction import logRecord
from apscheduler.schedulers.blocking import BlockingSchedulerdef getFileName(fileDir, fileExt):filenameList = []for root, dirs, files in os.walk(fileDir):for file in files:if os.path.splitext(file)[1] == '.{fileExt}'.format(fileExt=fileExt):filenameList.append(os.path.join(root, file))return filenameListdef execProgram_python_shell(scriptFile, fileExt):if fileExt == 'py':program = os.popen("which python").readline()[:-1]elif fileExt == 'sh':program = os.popen("which sh").readline()[:-1]print "[" + scriptFile + "] " + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f') + " exec start ...\n"os.system("{program} {scriptFile}".format(program=program, scriptFile=scriptFile))print "[" + scriptFile + "] " + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f') + " exec finished!\n"logRecord(scriptFile)def parallelExecProgram(fileDir, fileExt, parallelNum):parameterList = []requests = []for scriptFile in getFileName(fileDir, fileExt):parameterList.append(([scriptFile, fileExt], None))request_pyScript = threadpool.makeRequests(execProgram_python_shell, parameterList)requests.extend(request_pyScript)main_pool = threadpool.ThreadPool(parallelNum)[main_pool.putRequest(req) for req in requests]while True:try:time.sleep(1)main_pool.poll()except KeyboardInterrupt:print("**** Interrupted!")breakexcept threadpool.NoResultsPending:breakif main_pool.dismissedWorkers:print("Joining all dismissed worker threads...")main_pool.joinAllDismissedWorkers()def nowCronProgramMap(cronName):cronConfig = cronDirMap[cronName]cronStr = cronConfig['cron']fileDir = cronConfig['fileDir']fileExt = cronConfig['fileExt']parallelNum = cronConfig['parallelNum']cronDict = getCronList(conf_string=cronStr)nowTimeList = list(datetime.datetime.now().strftime('%M-%H-%d-%m-%w').split('-'))now_minute = int(nowTimeList[0])now_hour = int(nowTimeList[1])now_day = int(nowTimeList[2])now_month = int(nowTimeList[3])now_week = int(nowTimeList[4])if now_minute in cronDict['minute'] and now_hour in cronDict['hour'] and now_day in cronDict['day'] and now_month in \cronDict['month'] and now_week in cronDict['week']:parallelExecProgram(fileDir, fileExt, parallelNum)def schedJob():for cronName in cronDirMap.keys():if cronDirMap[cronName]['enable'] == "Ture":nowCronProgramMap(cronName=cronName)nowTime = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')print '当前时间: ' + nowTimeif __name__ == '__main__':sched = BlockingScheduler()sched.add_job(schedJob, 'interval', seconds=5)sched.start()

主脚本包含了一些脚本文件名称获取、python及shell执行外壳、并行处理模块及任务处理模块等。
6、并行的脚本
由于并行模块可能没有安装，所以将些脚本也拷进去。
/Users/nisj/PycharmProjects/BiDataProc/cronLake/threadpool.py

# -*- coding: UTF-8 -*-
"""Easy to use object-oriented thread pool framework.A thread pool is an object that maintains a pool of worker threads to perform
time consuming operations in parallel. It assigns jobs to the threads
by putting them in a work request queue, where they are picked up by the
next available thread. This then performs the requested operation in the
background and puts the results in another queue.The thread pool object can then collect the results from all threads from
this queue as soon as they become available or after all threads have
finished their work. It's also possible, to define callbacks to handle
each result as it comes in.The basic concept and some code was taken from the book "Python in a Nutshell,
2nd edition" by Alex Martelli, O'Reilly 2006, ISBN 0-596-10046-9, from section
14.5 "Threaded Program Architecture". I wrapped the main program logic in the
ThreadPool class, added the WorkRequest class and the callback system and
tweaked the code here and there. Kudos also to Florent Aide for the exception
handling mechanism.Basic usage::>>> pool = ThreadPool(poolsize)>>> requests = makeRequests(some_callable, list_of_args, callback)>>> [pool.putRequest(req) for req in requests]>>> pool.wait()See the end of the module code for a brief, annotated usage example.Website : http://chrisarndt.de/projects/threadpool/"""
__docformat__ = "restructuredtext en"__all__ = ['makeRequests','NoResultsPending','NoWorkersAvailable','ThreadPool','WorkRequest','WorkerThread'
]__author__ = "Christopher Arndt"
__version__ = '1.3.2'
__license__ = "MIT license"# standard library modules
import sys
import threading
import tracebacktry:import Queue            # Python 2
except ImportError:import queue as Queue   # Python 3# exceptions
class NoResultsPending(Exception):"""All work requests have been processed."""passclass NoWorkersAvailable(Exception):"""No worker threads available to process remaining requests."""pass# internal module helper functions
def _handle_thread_exception(request, exc_info):"""Default exception handler callback function.This just prints the exception info via ``traceback.print_exception``."""traceback.print_exception(*exc_info)# utility functions
def makeRequests(callable_, args_list, callback=None,exc_callback=_handle_thread_exception):"""Create several work requests for same callable with different arguments.Convenience function for creating several work requests for the samecallable where each invocation of the callable receives different valuesfor its arguments.``args_list`` contains the parameters for each invocation of callable.Each item in ``args_list`` should be either a 2-item tuple of the list ofpositional arguments and a dictionary of keyword arguments or a single,non-tuple argument.See docstring for ``WorkRequest`` for info on ``callback`` and``exc_callback``."""requests = []for item in args_list:if isinstance(item, tuple):requests.append(WorkRequest(callable_, item[0], item[1], callback=callback,exc_callback=exc_callback))else:requests.append(WorkRequest(callable_, [item], None, callback=callback,exc_callback=exc_callback))return requests# classes
class WorkerThread(threading.Thread):"""Background thread connected to the requests/results queues.A worker thread sits in the background and picks up work requests fromone queue and puts the results in another until it is dismissed."""def __init__(self, requests_queue, results_queue, poll_timeout=5, **kwds):"""Set up thread in daemonic mode and start it immediatedly.``requests_queue`` and ``results_queue`` are instances of``Queue.Queue`` passed by the ``ThreadPool`` class when it creates anew worker thread."""threading.Thread.__init__(self, **kwds)self.setDaemon(1)self._requests_queue = requests_queueself._results_queue = results_queueself._poll_timeout = poll_timeoutself._dismissed = threading.Event()self.start()def run(self):"""Repeatedly process the job queue until told to exit."""while True:if self._dismissed.isSet():# we are dismissed, break out of loopbreak# get next work request. If we don't get a new request from the# queue after self._poll_timout seconds, we jump to the start of# the while loop again, to give the thread a chance to exit.try:request = self._requests_queue.get(True, self._poll_timeout)except Queue.Empty:continueelse:if self._dismissed.isSet():# we are dismissed, put back request in queue and exit loopself._requests_queue.put(request)breaktry:result = request.callable(*request.args, **request.kwds)self._results_queue.put((request, result))except:request.exception = Trueself._results_queue.put((request, sys.exc_info()))def dismiss(self):"""Sets a flag to tell the thread to exit when done with current job."""self._dismissed.set()class WorkRequest:"""A request to execute a callable for putting in the request queue later.See the module function ``makeRequests`` for the common casewhere you want to build several ``WorkRequest`` objects for the samecallable but with different arguments for each call."""def __init__(self, callable_, args=None, kwds=None, requestID=None,callback=None, exc_callback=_handle_thread_exception):"""Create a work request for a callable and attach callbacks.A work request consists of the a callable to be executed by aworker thread, a list of positional arguments, a dictionaryof keyword arguments.A ``callback`` function can be specified, that is called when theresults of the request are picked up from the result queue. It mustaccept two anonymous arguments, the ``WorkRequest`` object and theresults of the callable, in that order. If you want to pass additionalinformation to the callback, just stick it on the request object.You can also give custom callback for when an exception occurs withthe ``exc_callback`` keyword parameter. It should also accept twoanonymous arguments, the ``WorkRequest`` and a tuple with the exceptiondetails as returned by ``sys.exc_info()``. The default implementationof this callback just prints the exception info via``traceback.print_exception``. If you want no exception handlercallback, just pass in ``None``.``requestID``, if given, must be hashable since it is used by``ThreadPool`` object to store the results of that work request in adictionary. It defaults to the return value of ``id(self)``."""if requestID is None:self.requestID = id(self)else:try:self.requestID = hash(requestID)except TypeError:raise TypeError("requestID must be hashable.")self.exception = Falseself.callback = callbackself.exc_callback = exc_callbackself.callable = callable_self.args = args or []self.kwds = kwds or {}def __str__(self):return "<WorkRequest id=%s args=%r kwargs=%r exception=%s>" % \(self.requestID, self.args, self.kwds, self.exception)class ThreadPool:"""A thread pool, distributing work requests and collecting results.See the module docstring for more information."""def __init__(self, num_workers, q_size=0, resq_size=0, poll_timeout=5):"""Set up the thread pool and start num_workers worker threads.``num_workers`` is the number of worker threads to start initially.If ``q_size > 0`` the size of the work *request queue* is limited andthe thread pool blocks when the queue is full and it tries to putmore work requests in it (see ``putRequest`` method), unless you alsouse a positive ``timeout`` value for ``putRequest``.If ``resq_size > 0`` the size of the *results queue* is limited and theworker threads will block when the queue is full and they try to putnew results in it... warning:If you set both ``q_size`` and ``resq_size`` to ``!= 0`` there isthe possibilty of a deadlock, when the results queue is not pulledregularly and too many jobs are put in the work requests queue.To prevent this, always set ``timeout > 0`` when calling``ThreadPool.putRequest()`` and catch ``Queue.Full`` exceptions."""self._requests_queue = Queue.Queue(q_size)self._results_queue = Queue.Queue(resq_size)self.workers = []self.dismissedWorkers = []self.workRequests = {}self.createWorkers(num_workers, poll_timeout)def createWorkers(self, num_workers, poll_timeout=5):"""Add num_workers worker threads to the pool.``poll_timout`` sets the interval in seconds (int or float) for howofte threads should check whether they are dismissed, while waiting forrequests."""for i in range(num_workers):self.workers.append(WorkerThread(self._requests_queue,self._results_queue, poll_timeout=poll_timeout))def dismissWorkers(self, num_workers, do_join=False):"""Tell num_workers worker threads to quit after their current task."""dismiss_list = []for i in range(min(num_workers, len(self.workers))):worker = self.workers.pop()worker.dismiss()dismiss_list.append(worker)if do_join:for worker in dismiss_list:worker.join()else:self.dismissedWorkers.extend(dismiss_list)def joinAllDismissedWorkers(self):"""Perform Thread.join() on all worker threads that have been dismissed."""for worker in self.dismissedWorkers:worker.join()self.dismissedWorkers = []def putRequest(self, request, block=True, timeout=None):"""Put work request into work queue and save its id for later."""assert isinstance(request, WorkRequest)# don't reuse old work requestsassert not getattr(request, 'exception', None)self._requests_queue.put(request, block, timeout)self.workRequests[request.requestID] = requestdef poll(self, block=False):"""Process any new results in the queue."""while True:# still results pending?if not self.workRequests:raise NoResultsPending# are there still workers to process remaining requests?elif block and not self.workers:raise NoWorkersAvailabletry:# get back next resultsrequest, result = self._results_queue.get(block=block)# has an exception occured?if request.exception and request.exc_callback:request.exc_callback(request, result)# hand results to callback, if anyif request.callback and not \(request.exception and request.exc_callback):request.callback(request, result)del self.workRequests[request.requestID]except Queue.Empty:breakdef wait(self):"""Wait for results, blocking until all have arrived."""while 1:try:self.poll(True)except NoResultsPending:break################
# USAGE EXAMPLE
################if __name__ == '__main__':import randomimport time# the work the threads will have to do (rather trivial in our example)def do_something(data):time.sleep(random.randint(1,5))result = round(random.random() * data, 5)# just to show off, we throw an exception once in a whileif result > 5:raise RuntimeError("Something extraordinary happened!")return result# this will be called each time a result is availabledef print_result(request, result):print("**** Result from request #%s: %r" % (request.requestID, result))# this will be called when an exception occurs within a thread# this example exception handler does little more than the default handlerdef handle_exception(request, exc_info):if not isinstance(exc_info, tuple):# Something is seriously wrong...print(request)print(exc_info)raise SystemExitprint("**** Exception occured in request #%s: %s" % \(request.requestID, exc_info))# assemble the arguments for each job to a list...data = [random.randint(1,10) for i in range(20)]# ... and build a WorkRequest object for each item in datarequests = makeRequests(do_something, data, print_result, handle_exception)# to use the default exception handler, uncomment next line and comment out# the preceding one.#requests = makeRequests(do_something, data, print_result)# or the other form of args_lists accepted by makeRequests: ((,), {})data = [((random.randint(1,10),), {}) for i in range(20)]requests.extend(makeRequests(do_something, data, print_result, handle_exception)#makeRequests(do_something, data, print_result)# to use the default exception handler, uncomment next line and comment# out the preceding one.)# we create a pool of 3 worker threadsprint("Creating thread pool with 3 worker threads.")main = ThreadPool(3)# then we put the work requests in the queue...for req in requests:main.putRequest(req)print("Work request #%s added." % req.requestID)# or shorter:# [main.putRequest(req) for req in requests]# ...and wait for the results to arrive in the result queue# by using ThreadPool.wait(). This would block until results for# all work requests have arrived:# main.wait()# instead we can poll for results while doing something else:i = 0while True:try:time.sleep(0.5)main.poll()print("Main thread working...")print("(active worker threads: %i)" % (threading.activeCount()-1, ))if i == 10:print("**** Adding 3 more worker threads...")main.createWorkers(3)if i == 20:print("**** Dismissing 2 worker threads...")main.dismissWorkers(2)i += 1except KeyboardInterrupt:print("**** Interrupted!")breakexcept NoResultsPending:print("**** No pending results.")breakif main.dismissedWorkers:print("Joining all dismissed worker threads...")main.joinAllDismissedWorkers()