celery 和 haystack

celery  是分布式异步框架 

haystack  是全文检索  只能在Django中用。

 

 

1、什么是celery？     ---->它是Python写的，因此只支持Python使用。可是消息队列是通用的任何语言均可以用。

　　celery英文翻译是芹菜。

　　celery是一个简单、灵活且可靠的、处理大量消息的分布式系统。

　　专一于实时处理的异步任务队列。

　　同时也支持任务调度。

celery 架构  由消息中间件broker、任务执行单元worker、任务执行结果存储task result store，这三部分组成。

 

user表明我们程序员写的代码，我们提交一些执行的任务，好比说发个邮件。---提交到celery框架--->amqp broker 消息队列----->celery worker 具体执行------>task result store 结果的存储。

用户提交一些任务，任务就是一些函数-----提交到消息队列，排队等着去执行。   生产者消费者模型

amqp broker 消息队列------>消息中间件--->celery自己不提供消息服务，可是能够方便的和第三方提供的消息中间件组成。包括。rabbitMQ，Redis（能够作数据库，能够作缓存，能够作数据队列...咱们会详细介绍这个redis）等等

celery worker 具体执行------>任务执行单元--->worker是celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。

task result store 结果的存储------>任务执行存储结果存储--->用来存储worker执行的任务的结果，celery支持以不一样方式存储任务的结果，包括AMOB，redis等。

 

什么是消息队列？

什么是异步处理？

 

Celery version 4.0 runs on
        Python ❨2.7, 3.4, 3.5❩
        PyPy ❨5.4, 5.5❩
    This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required.

    If you’re running an older version of Python, you need to be running an older version of Celery:

        Python 2.6: Celery series 3.1 or earlier.
        Python 2.5: Celery series 3.0 or earlier.
        Python 2.4 was Celery series 2.2 or earlier.

    Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.

celery 版本支持状况　

2、使用场景

　　异步任务：将耗时操做任务提交给celery去异步执行，好比发送短信/邮件、消息推送、音视频处理等等

　　定时任务：定时执行某件事情，好比天天数据统计。

 

安装

pip源或者pip3 install celery == 版本号

 

 

 

3、基本使用

        -1.生成一个celery对象(传参数),cel.task装饰器,装饰函数(这个函数就能够被提交到任务队列中)
        -2.把函数提交到任务队列中 函数.delay(参数)
        -3.启动worker:celery worker -A celery_task -l info -P eventlet
        -4.查询任务执行结果

-简单使用

一、首先须要生成celery对象（传参数），并写个任务（函数）---->提交任务---->查询任务结果---->开启worker---->查询结果

建立项目celerytest

建立py文件：celery_app_task.py


import celery
import time
# broker='redis://127.0.0.1:6379/2' 不加密码
backend='redis://:123456@127.0.0.1:6379/1'
broker='redis://:123456@127.0.0.1:6379/2'
cel=celery.Celery('test',backend=backend,broker=broker)
@cel.task
def add(x,y):
    return x+y

 

建立py文件：add_task.py,添加任务

from celery_app_task import add
result = add.delay(4,5)
print(result.id)

 

建立py文件：run.py，执行任务，或者使用命令执行：celery worker -A celery_app_task -l info

注：windows下：celery worker -A celery_app_task -l info -P eventlet

from celery_app_task import cel
if __name__ == '__main__':
    cel.worker_main()
    # cel.worker_main(argv=['--loglevel=info')

 

rom celery.result import AsyncResult
from celery_app_task import cel

async = AsyncResult(id="e919d97d-2938-4d0f-9265-fd8237dc2aa3", app=cel)

if async.successful():
    result = async.get()
    print(result)
    # result.forget() # 将结果删除
elif async.failed():
    print('执行失败')
elif async.status == 'PENDING':
    print('任务等待中被执行')
elif async.status == 'RETRY':
    print('任务异常后正在重试')
elif async.status == 'STARTED':
    print('任务已经开始被执行')

 

 

4、多任务结构

 pro_cel
    ├── celery_task# celery相关文件夹
    │   ├── celery.py   # celery链接和配置相关文件,必须叫这个名字
    │   └── tasks1.py    #  全部任务函数
    │    └── tasks2.py    #  全部任务函数
    ├── check_result.py # 检查结果
    └── send_task.py    # 触发任务

 

from celery import Celery

cel = Celery('celery_demo',
             broker='redis://127.0.0.1:6379/1',
             backend='redis://127.0.0.1:6379/2',
             # 包含如下两个任务文件，去相应的py文件中找任务，对多个任务作分类
             include=['celery_task.tasks1',
                      'celery_task.tasks2'
                      ])

# 时区
cel.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
cel.conf.enable_utc = False

celery.py

import time
from celery_task.celery import cel

@cel.task
def test_celery(res):
    time.sleep(5)
    return "test_celery任务结果:%s"%res

task1.py

import time
from celery_task.celery import cel
@cel.task
def test_celery2(res):
    time.sleep(5)
    return "test_celery2任务结果:%s"%res

task2.py

from celery.result import AsyncResult
from celery_task.celery import cel

async = AsyncResult(id="08eb2778-24e1-44e4-a54b-56990b3519ef", app=cel)

if async.successful():
    result = async.get()
    print(result)
    # result.forget() # 将结果删除,执行完成，结果不会自动删除
    # async.revoke(terminate=True)  # 不管如今是何时，都要终止
    # async.revoke(terminate=False) # 若是任务尚未开始执行呢，那么就能够终止。
elif async.failed():
    print('执行失败')
elif async.status == 'PENDING':
    print('任务等待中被执行')
elif async.status == 'RETRY':
    print('任务异常后正在重试')
elif async.status == 'STARTED':
    print('任务已经开始被执行')

check_result.py

from celery_task.tasks1 import test_celery
from celery_task.tasks2 import test_celery2

# 当即告知celery去执行test_celery任务，并传入一个参数
result = test_celery.delay('第一个的执行')
print(result.id)
result = test_celery2.delay('第二个的执行')
print(result.id)

send_task.py

添加任务（执行send_task.py），开启work：celery worker -A celery_task(这是文件夹) -l info  -P  eventlet，检查任务执行结果（执行check_result.py）

 

 

5、设定延时时间让celery执行一个任务

from celery_app_task import add
from datetime import datetime
eta 时间对象为打他time类型
# 方式一
#经过这个方式拿到一个时间对象，就是参数传的时间
# v1 = datetime(2019, 2, 13, 18, 19, 56)     #2019年2月13号18点19分56秒
# print(v1)
#把当地时间转成UTC时间
# v2 = datetime.utcfromtimestamp(v1.timestamp())
# print(v2)
# result = add.apply_async(args=[1, 3], eta=v2)
# print(result.id)

# 方式二
ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
#取一个10秒以后的时间
time_delay = timedelta(seconds=10)
#这个时间是当前时间日后推10秒
task_time = utc_ctime + time_delay

# 使用apply_async并设定时间
result = add.apply_async(args=[4, 3], eta=task_time)
print(result.id)

add_task.py

6、设定时间让celery执行一个任务

 相似于contab的定时任务

多任务结构中celery.py修改以下

from datetime import timedelta
from celery import Celery
from celery.schedules import crontab

cel = Celery('tasks', broker='redis://127.0.0.1:6379/1', backend='redis://127.0.0.1:6379/2', include=[
    'celery_task.tasks1',
    'celery_task.tasks2',
])
cel.conf.timezone = 'Asia/Shanghai'
cel.conf.enable_utc = False

cel.conf.beat_schedule = {
    # 名字随意命名
    'add-every-10-seconds': {
        # 执行tasks1下的test_celery函数
        'task': 'celery_task.tasks1.test_celery',
        # 每隔2秒执行一次
        # 'schedule': 2.0,
        # 'schedule': crontab(minute="*/1"),
        'schedule': timedelta(seconds=2),
        # 传递参数
        'args': ('test',)
    },
    # 'add-every-12-seconds': {
    #     'task': 'celery_task.tasks2.test_celery2',
    #     每一年4月11号，8点42分执行
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'args': (‘test2’，)
    # },
}

第一步：在celery.py里面加：

from datetime import timedelta
from celery.schedules import crontab
cel.conf.beat_schedule = {
    # 名字随意命名
    'add-every-10-seconds': {
        # 执行tasks1下的test_celery函数
        'task': 'celery_task.tasks1.test_celery',
        # 每隔2秒执行一次 
        # 'schedule': 1.0,
        # 'schedule': crontab(minute="*/1"),
        'schedule': timedelta(seconds=2),
        # 传递参数
        'args': ('test',)
    },
    # 'add-every-12-seconds': {
    #     'task': 'celery_task.tasks1.test_celery',
    #     每一年4月11号，8点42分执行
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'args': (16, 16)
    # },
}

加的内容

第二步：启动一个beat：celery beat -A celery_task -l info     ---》启动一个beat往队列里提交任务

第三步：启动work执行：celery worker -A celery_task -l info -P eventlet      ----》启动worker

 

 

 

在Django中使用celery

Django-celery版本的要求很是苛刻 不只跟celery版本有关系还跟Django版本有关系。

 

怎么使用？  也须要本身启动worker

第一步：在项目根目录下建立celeryconfig.py

import djcelery
djcelery.setup_loader()
CELERY_IMPORTS=(
    'app01.tasks',
)
#有些状况能够防止死锁
CELERYD_FORCE_EXECV=True
# 设置并发worker数量
CELERYD_CONCURRENCY=4
#容许重试
CELERY_ACKS_LATE=True
# 每一个worker最多执行100个任务被销毁，能够防止内存泄漏
CELERYD_MAX_TASKS_PER_CHILD=100
# 超时时间
CELERYD_TASK_TIME_LIMIT=12*30

celeryconfig.py

第二步：在app01目录下建立task.py

from celery import task
@task
def add(a,b):
    with open('a.text', 'a', encoding='utf-8') as f:
        f.write('a')
    print(a+b)

View Code

第三步：视图函数view.py 

from django.shortcuts import render,HttpResponse
from app01.tasks import add
from datetime import datetime
def test(request):
    # result=add.delay(2,3)
    ctime = datetime.now()
    # 默认用utc时间
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    from datetime import timedelta
    time_delay = timedelta(seconds=5)
    task_time = utc_ctime + time_delay
    result = add.apply_async(args=[4, 3], eta=task_time)
    print(result.id)
    return HttpResponse('ok')

View Code

第四步：setting.py中配置

INSTALLED_APPS = [
    ...
    'djcelery',
    'app01'
]

...

from djagocele import celeryconfig
BROKER_BACKEND='redis'
BOOKER_URL='redis://127.0.0.1:6379/1'
CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/2

View Code

 

启动worker

python3 manage.py celery worker 

 

 

 

haystack

一、什么是haystack?

　　是Django的开源全文搜索框架。底层依附于solr、elasticsearch、whoosh...这些库　　

- 全文检索不一样于特定字段的模糊查询，使用全文检索的效率更高，而且可以对于中文进行分词处理
- haystack：django的一个包，能够方便地对model里面的内容进行索引、搜索，设计为支持whoosh,solr,Xapian,Elasticsearc四种全文检索引擎后端，属于一种全文检索的框架
- whoosh：纯Python编写的全文搜索引擎，虽然性能比不上sphinx、xapian、Elasticsearc等，可是无二进制包，程序不会莫名其妙的崩溃，对于小型的站点，whoosh已经足够使用
- jieba：一款免费的中文分词包，若是以为很差用可使用一些收费产品

 

 

二、安装：

 

pip install django-haystack
pip install whoosh
pip install jieba

 

 

三、配置

添加Haystack到INSTALLED_APPS 跟大多数Django的应用同样，你应该在你的设置文件(一般是settings.py)添加Haystack到INSTALLED_APPS. 示例：

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.sites',

    # 添加
    'haystack',

    # 你的app
    'blog',
]

 

修改settings.py 在你的settings.py中，你须要添加一个设置来指示站点配置文件正在使用的后端，以及其它的后端设置。 HAYSTACK——CONNECTIONS是必需的设置，而且应该至少是如下的一种：

#须要设置PATH到你的Whoosh索引的文件系统位置
import os
HAYSTACK_CONNECTIONS = {
    'default': {
        'ENGINE': 'haystack.backends.whoosh_backend.WhooshEngine',
        'PATH': os.path.join(os.path.dirname(__file__), 'whoosh_index'),
    },
}

# 自动更新索引
HAYSTACK_SIGNAL_PROCESSOR = 'haystack.signals.RealtimeSignalProcessor'

whoosh示例

 

 

四、处理数据

建立索引