---

前言

Windows 系統進行 Django 開發工作，然后原來使用的django-crontab插件沒辦法在Windows系統上面進行定時任務。因此又想了其他方式來實現定時任務。下面就來說說這些方案的優缺點。

---

使用django-crontab插件來實現定時任務

安裝庫

pip install django-crontab

注冊app

在settings.py中的 INSTALLED_APPS注冊app

INSTALLED_APPS = [
    ...
    'django_crontab'
]

在settings.py中配置定時任務

"""
# 元素的第一個參數是 頻次(多久一次)
*  *  *  * *
分 時 日 月 周    命令

M: 分鐘（0-59）。每分鐘用 * 或者 */1 表示
H：小時（0-23）。（0表示0點）
D：天（1-31）。
m: 月（1-12）。
d: 一星期內的天（0~6，0為星期天）。

*/5 * * * * ---- 每5分鐘執行一次

# 元素的第二個參數是 定時任務（函數）

finally run this command to add all defined jobs from CRONJOBS to crontab (of the user which you are running this command with):
在終端運行: python manage.py crontab add

show current active jobs of this project:
在終端運行: python manage.py crontab show

removing all defined jobs is straight forward:
在終端運行: python manage.py crontab remove
"""
crontab 只在Linux下才能使用
CRONJOBS = [
    ('*/5 * * * *', 'myapp.cron.my_scheduled_job', ' >> /tmp/logs/confdict_handle.log')
]

編寫定時任務方法

在子應用新建 crontabs.py 文件，寫入定時任務

import datetime

# 定時任務
def confdict_handle():
    try:
        loca_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        print('本地時間：'+str(loca_time))
    except Exception as e:
        print('發生錯誤，錯誤信息為：', e)

使用&運行

開啟定時器

python manage.py crontab add

查看開啟的定時器

python manage.py crontab show

關閉定時器

python manage.py crontab remove

優缺點

優點：
簡單、方便、易于管理和django服務是分離的，不會影響到django對外提供的web服務。
缺點：
無法在Windows平臺上運行
官方文檔：

就算在Linux系統上，也可能出現運行了沒有效果的消息，至今未知原因

使用django-apscheduler插件實現定時任務

安裝庫

pip install django-apscheduler

注冊app

修改settings.py

INSTALLED_APPS = (
  ...
  "django_apscheduler",
)

遷移數據庫

因為django-apscheduler會創建表來存儲定時任務的一些信息，所以將app加入之后需要遷移數據

python manage.py migrate

完整示例 在views.py中增加你的定時任務代碼

from apscheduler.schedulers.background import BackgroundScheduler # 使用它可以使你的定時任務在后臺運行
from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job
import time
'''
date：在您希望在某個特定時間僅運行一次作業時使用
interval：當您要以固定的時間間隔運行作業時使用
cron：以crontab的方式運行定時任務
minutes：設置以分鐘為單位的定時器
seconds：設置以秒為單位的定時器
'''

try:
    scheduler = BackgroundScheduler()
    scheduler.add_jobstore(DjangoJobStore(), "default")

    @register_job(scheduler, "interval", seconds=5)
    def test_job():
        # 定時每5秒執行一次
        print(time.strftime('%Y-%m-%d %H:%M:%S'))

    register_events(scheduler)
    # 啟動定時器
    scheduler.start()
except Exception as e:
    print('定時任務異常：%s' % str(e))

使用&運行

apscheduler定時任務會跟隨django項目一起運行因此直接啟動django即可

python manage.py runserver

詳細查看另一篇文章: django-apscheduler

優缺點

優點：
簡單、定時方式豐富、輕量
缺點：
定時任務會跟隨django項目一起運行，會影響django對外提供的web服務
使用uwsgi線上運行時，很難啟動apscheduler定時任務
按照文檔上的設置–enable-threads依然沒有正常啟動，具體原因未知，也查了很多方法，都沒有解決。并且也不建議將定時任務和uwsgi放在一起運行，這樣定時任務在運行過程中可能會影響uwsgi的服務。

使用Celery插件實現定時任務

介紹

"""
生產者（任務，函數）
    @app.task
    def celery_send_sms_code(mobie,code):

        CCP().send_template_sms(mobie,[code,5],1)


    app.autodiscover_tasks(['celery_tasks.sms'])
消費者
    celery -A proj worker -l INFO

    在虛擬環境下執行指令
    celery -A celery實例的腳本路徑 worker -l INFO

    本案例使用
    celery -A celery_tasks.main worker -l INFO

隊列（中間人、經紀人）
    #2. 設置broker
    # 我們通過加載配置文件來設置broker
    app.config_from_object('celery_tasks.config')

    # 配置信息 key=value
    # 我們指定 redis為我們的broker(中間人，經紀人，隊列)
    broker_url="redis://127.0.0.0:6379/15"

Celery -- 將這3者實現了。

    # 0. 為celery的運行 設置Django的環境
    import os
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'meiduo_mall.settings')

    # 1. 創建celery實例
    from celery import Celery
    # 參數1： main 設置腳本路徑就可以了。 腳本路徑是唯一的
    app=Celery('celery_tasks')
"""

安裝庫

本例建立在認為你已經知道如何使用Celery實現異步任務的基礎上，需要學習的請移步Django使用Celery,可參考另一篇博客：celery異步—生產者消費者，或者官方文檔：官方文檔
本例使用redis作為Borker和backend

pip install -U "celery[redis]"

配置celery

settings.py 增加以下代碼

# Celery配置
from kombu import Exchange, Queue
# 設置任務接受的類型，默認是{'json'}
CELERY_ACCEPT_CONTENT = ['application/json']
# 設置task任務序列列化為json
CELERY_TASK_SERIALIZER = 'json'
# 請任務接受后存儲時的類型
CELERY_RESULT_SERIALIZER = 'json'
# 時間格式化為中國時間
CELERY_TIMEZONE = 'Asia/Shanghai'
# 是否使用UTC時間
CELERY_ENABLE_UTC = False
# 指定borker為redis 如果指定rabbitmq CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'
CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
# 指定存儲結果的地方，支持使用rpc、數據庫、redis等等，具體可參考文檔 # CELERY_RESULT_BACKEND = 'db+mysql://scott:tiger@localhost/foo' # mysql 作為后端數據庫
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'
# 設置任務過期時間 默認是一天，為None或0 表示永不過期
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
# 設置worker并發數，默認是cpu核心數
# CELERYD_CONCURRENCY = 12
# 設置每個worker最大任務數
CELERYD_MAX_TASKS_PER_CHILD = 100
# 指定任務的位置
CELERY_IMPORTS = (
    'base.tasks',
)
# 使用beat啟動Celery定時任務
# schedule時間的具體設定參考：https://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html
CELERYBEAT_SCHEDULE = {
    'add-every-10-seconds': {
        'task': 'base.tasks.cheduler_task',
        'schedule': 10,
        'args': ('hello', )
    },
}

編寫定時任務代碼

在本例中是在子應用新建 tasks.py 在裏面增加的定時任務

# Create your tasks here
from __future__ import absolute_import, unicode_literals
from celery import shared_task
import time

@shared_task
def cheduler_task(name):
    print(name)
    print(time.strftime('%X'))

使用&運行

啟動定時任務

celery -A dase_django_api beat -l info

啟動Celery worker 用來執行定時任務

celery -A dase_django_api worker -l -l info

注意：這里的 dase_django_api 要換成你的Celery APP的名稱

優缺點

優點：
穩定可靠、管理方便
高性能、支持分布式
缺點：
實現復雜
部署復雜
非輕量級

自建代碼實現定時任務

創建定時任務

在子應用下創建一個文件名為 schedules.py，鍵入以下內容

import os, sys, time, datetime
import threading
import django
base_apth = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# print(base_apth)
# 將項目路徑加入到系統path中，這樣在導入模型等模塊時就不會報模塊找不到了
sys.path.append(base_apth)
os.environ['DJANGO_SETTINGS_MODULE'] ='base_django_api.settings' # 注意：base_django_api 是我的模塊名，你在使用時需要跟換為你的模塊
django.setup()
from base.models import ConfDict

def confdict_handle():
    while True:
        try:
            loca_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            print('本地時間：'+str(loca_time))
            time.sleep(10)
        except Exception as e:
            print('發生錯誤，錯誤信息為：', e)
            continue


def main():
    '''
    主函數，用于啟動所有定時任務，因為當前定時任務是手動實現，因此可以自由發揮
    '''
    try:
        # 啟動定時任務，多個任務時，使用多線程
        task1 = threading.Thread(target=confdict_handle)
        task1.start()
    except Exception as e:
        print('發生異常：%s' % str(e))

if __name__ == '__main__':
    main()

使用&運行

直接運行腳本即可

python apps/base/schedules.py

注意：apps/base 為你的子應用路徑

優缺點

優點：
自定義
高度自由
缺點：
過于簡單
當任務過多時，占用資源也會增加

---