目的：很多時(shí)候，我們需要多個(gè)不同策略去完成一個(gè)任務(wù)，那個(gè)插件架構(gòu)是很好一個(gè)選擇。而每一個(gè)策略，不希望一個(gè)個(gè)去執(zhí)行，還能實(shí)時(shí)上報(bào)數(shù)據(jù)，所以每一個(gè)插件都是異步執(zhí)行，把每一個(gè)插件結(jié)果回調(diào)方法實(shí)時(shí)上報(bào)，然后寫入數(shù)據(jù)。

?

線程工具類

#app/MyThread.py

            
              import threading
class MyThread(threading.Thread):
    def __init__(self, func, **kwargs):
        threading.Thread.__init__(self)
        self.func = func
        self.kwargs = kwargs
        self.thread_stop = False

    def run(self):
        self.func(self, self.kwargs)

    def stop(self):
        self.thread_stop = True
            
          

插件管理中心

#app/platform.py

            
              #coding=utf-8
from .MyThread import MyThread
# 注冊(cè)加載插件
# 多線程運(yùn)行插件
# 回調(diào)返回?cái)?shù)據(jù)，放入寫入隊(duì)列
class DataManagerProcessor(object):
    PLUGINS = {}

    def run_plugin(self,parent,args):
        self.PLUGINS[args["plugin_name"]]().run(parent,args["callback"])

    def process(self, callback):
        threads=[]
        for plugin_name in self.PLUGINS.keys():
            print(plugin_name)
            dt = MyThread(self.run_plugin,plugin_name=plugin_name,callback=callback)
            dt.start()
            dt.setName(plugin_name)
            threads.append(dt)
        return threads

    @classmethod
    def plugin_register(cls, plugin_name):
        def wrapper(plugin):
            cls.PLUGINS.update({plugin_name:plugin})
            return plugin
        return wrapper
            
          

#app/main.py

            
              #coding=utf-8

import queue

from .MyThread import MyThread
from .platform import DataManagerProcessor

result_queue = queue.Queue()

#異步寫入隊(duì)列
def updateDB(result):
    result_queue.put(result)

#判斷插件任務(wù)線程是否結(jié)束
def isAllTasksCompleted(threads):
    for t in threads:
        if t.is_alive():
            return False
    return True

def syncDB(parent,args):
    threads=args["threads"]
    while True:
        if not result_queue.empty():
            data=result_queue.get()
            print("*************************")
            print(data)
            print("*************************")
            result_queue.task_done()
        elif isAllTasksCompleted(threads):
            break


def main():
    processor = DataManagerProcessor()
    threads=processor.process(updateDB)
    # for t in threads:
    #     t.join()

    syncThread=MyThread(syncDB,threads=threads)
    syncThread.start()
    syncThread.join()


    print("Done")
            
          

#app/__init__.py

            
              from .plugins import *
            
          

#run.py

            
              from app.main import main

main()
            
          

插件編寫

#app/plugins/plugin1.py

            
              from app.platform import DataManagerProcessor

@DataManagerProcessor.plugin_register('plugin1')
class plugin1(object):
   
    def crawl_data(self,callback):
        callback(1)#實(shí)時(shí)上報(bào)數(shù)據(jù)

    #parent 是線程對(duì)象，可以隨時(shí)停止任務(wù)
    def run(self, parent,callback):
        print(parent.thread_stop)
        self.crawl_data(callback)
        
        

if __name__ == '__main__':
    def test(result):
        print(result)
    plugin1().crawl_data(result)
            
          

#app/plugins/plugin2.py

            
              from app.platform import DataManagerProcessor

@DataManagerProcessor.plugin_register('plugin1')
class plugin2(object):
   
    def crawl_data(self,callback):
        callback(2)#實(shí)時(shí)上報(bào)數(shù)據(jù)

    #parent 是線程對(duì)象，可以隨時(shí)停止任務(wù)
    def run(self, parent,callback):
        print(parent.thread_stop)
        self.crawl_data(callback)
        
        

if __name__ == '__main__':
    def test(result):
        print(result)
    plugin2().crawl_data(result)
            
          

#app/plugins/__init__.py

            
              __all__ = ['plugin1', 'plugin2']
            
          

?