python(08)实践Django-异步任务celery

本项目使用celery 异步去同步项目在jenkins中的构建信息。

最终结果展示：

1. 在项目的构建列表中点击同步按钮，调用celery去jenkins同步所有构建信息

   
   Paste_Image.png

2. 每隔12小时，调用celery的定时任务去jenkins同步最新build信息

   
   Paste_Image.png

前提：已有redis服务并启动

1. 安装celery

     pip install celery
     pip install redis
   

2. settings.py 配置

    BROKER_URL = 'redis://ip:6379/6'
    CELERY_RESULT_BACKEND = BROKER_URL
    CELERY_TIMEZONE = 'Asia/Shanghai'
    #时区不一致时会导致任务监控不到
   

3. init.py

   from .celery import app as celery_app
   

4. celery.py 配置

    # ~*~ coding: utf-8 ~*~
    from __future__ import absolute_import, unicode_literals
    import os
    from celery import Celery
    from celery.schedules import crontab
    from django.conf import settings
   
    # set the default Django settings module for the 'celery' program.
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Testmanager.settings')
    app = Celery('Testmanager')
   
    # Using a string here means the worker will not have to
    # pickle the object when using Windows.
    app.config_from_object('django.conf:settings')
    app.autodiscover_tasks(lambda: [app_config.split('.')[0]
                                for app_config in settings.INSTALLED_APPS])
     #定时任务配置
     app.conf.update(
    CELERYBEAT_SCHEDULE={
       'refresh-pci-last-build-info': {
            'task': 'devops.tasks.update_all_pcis_by_jenkins',
           'schedule': crontab(hour='*/12'),
            'args': (),
            }
        }
    )
   

5. task任务

   @shared_task
   def update_all_pcis_by_jenkins():
    jen = jenkins.Jenkins(jenkins_url, username=jenkins_user, password=jenkins_passwd)
    pcis = PCI.objects.all()
    for pci in pcis :
        job_name = pci.job_name
        lastBuild_Number = jen.get_job_info(job_name)['lastBuild']['number']
        if pci.last_build_number != lastBuild_Number :
            lastBuild_url = jen.get_build_info(job_name, lastBuild_Number)['url']  # 最后一次build号
            times = jen.get_build_info(job_name, lastBuild_Number)['timestamp']  # build的日期
            result = jen.get_build_info(job_name, lastBuild_Number)['result']
            pci.last_build_number = lastBuild_Number
            pci.last_build_result = result
            pci.last_build_url = lastBuild_url
            pci.last_update_time = timestamp2datetime(times)
            pci.save()
        print(job_name)
        print('已是最新版本')
   @shared_task
   def update_pcis_by_jenkins(job_name,pid):
     jen = jenkins.Jenkins(jenkins_url, username=jenkins_user, password=jenkins_passwd)
     print (job_name)
     pci = PCI.objects.get(id=pid)
     job_info_scm = jen.get_job_info(job_name)['scm']['_class']
    #{'_class': 'hudson.plugins.git.GitSCM'}
     print('job_info_scm')
     print(job_info_scm)
     if job_info_scm.find('git') != -1:
         pci.pro_scm = 'git'
    else:
        pci.pro_scm = 'svn'
     lastBuild_Number = jen.get_job_info(job_name)['lastBuild']['number']
     lastBuild_url = jen.get_build_info(job_name, lastBuild_Number)['url']# 最后一次build号
     times = jen.get_build_info(job_name, lastBuild_Number)['timestamp']# build的日期
     result = jen.get_build_info(job_name, lastBuild_Number)['result']
     pci.last_build_number = lastBuild_Number
     pci.last_build_result = result
     pci.last_build_url = lastBuild_url
     pci.last_update_time = timestamp2datetime(times)
     pci.save()
   

6. 启动celery 进程

   celery -A Testmanager worker --loglevel=info
   #定时任务
   celery -A Testmanager beat --loglevel=info
   

加载已有的tasks
定时任务