celery haystack

Posted 2021-02-17 wrqysrt

# Celery

## 1.什么是Celery

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

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

同时也支持任务调度

### Celery架构


Celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。

#### 消息中间件

Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等

#### 任务执行单元

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。

#### 任务结果存储

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, 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.
```

 

## 2.使用场景

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

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

## 3.Celery的安装配置

pip install celery

消息中间件：RabbitMQ/Redis

app=Celery(‘任务名‘，backend=‘xxx‘,broker=‘xxx‘)

## 4.Celery执行异步任务

#### 基本使用

创建项目celerytest

创建py文件：celery_app_task.py

```python
import celery
import time
# broker=‘redis://127.0.0.1:6379/2‘ 不加密码
backend=‘redis://:[email protected]:6379/1‘
broker=‘redis://:[email protected]:6379/2‘
cel=celery.Celery(‘test‘,backend=backend,broker=broker)
@cel.task
def add(x,y):
return x+y

```

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

```python
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

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

创建py文件：result.py，查看任务执行结果

```python
from 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(‘任务已经开始被执行‘)
```

执行 add_task.py，添加任务，并获取任务ID

执行 run.py ，或者执行命令：celery worker -A celery_app_task -l info

执行 result.py,检查任务状态并获取结果

#### 多任务结构

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

celery.py

```python
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
```

tasks1.py

```python
import time
from celery_task.celery import cel

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

tasks2.py

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

check_result.py

```python
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(‘任务已经开始被执行‘)
```

send_task.py

```python
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），开启work：celery worker -A celery_task -l info -P eventlet，检查任务执行结果（执行check_result.py）

## 5.Celery执行定时任务

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

add_task.py

```python
from celery_app_task import add
from datetime import datetime

# 方式一
# v1 = datetime(2019, 2, 13, 18, 19, 56)
# print(v1)
# 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
time_delay = timedelta(seconds=10)
task_time = utc_ctime + time_delay

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

#### 类似于contab的定时任务

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

```python
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‘: 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

启动work执行：celery worker -A celery_task -l info -P eventlet

## 6.Django中使用Celery

安装包

```python
celery==3.1.25
django-celery==3.1.20
```

 

在项目目录下创建celeryconfig.py

```python
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
```

在app01目录下创建tasks.py

```python
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)
```

视图函数views.py

```python
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‘)
```

settings.py

```python

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‘
```

 

 

 

 

 

 

 

 

 

# Haystack

## 1.什么是Haystack

Haystack是django的开源全文搜索框架(全文检索不同于特定字段的模糊查询，使用全文检索的效率更高 )，该框架支持**Solr**,**Elasticsearch**,**Whoosh**, ***Xapian*搜索引擎它是一个可插拔的后端（很像Django的数据库层），所以几乎你所有写的代码都可以在不同搜索引擎之间便捷切换

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

## 2.安装

```python
pip install django-haystack
pip install whoosh
pip install jieba
```

## 3.配置

###添加Haystack到`INSTALLED_APPS`

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

```python
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`是必需的设置，并且应该至少是以下的一种：

#### Solr示例

```python
HAYSTACK_CONNECTIONS = {
‘default‘: {
‘ENGINE‘: ‘haystack.backends.solr_backend.SolrEngine‘,
‘URL‘: ‘http://127.0.0.1:8983/solr‘
# ...or for multicore...
# ‘URL‘: ‘http://127.0.0.1:8983/solr/mysite‘,
},
}
```

#### Elasticsearch示例

```python
HAYSTACK_CONNECTIONS = {
‘default‘: {
‘ENGINE‘: ‘haystack.backends.elasticsearch_backend.ElasticsearchSearchEngine‘,
‘URL‘: ‘http://127.0.0.1:9200/‘,
‘INDEX_NAME‘: ‘haystack‘,
},
}
```

#### Whoosh示例

```python
#需要设置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‘
```

#### Xapian示例

```python
#首先安装Xapian后端（http://github.com/notanumber/xapian-haystack/tree/master）
#需要设置PATH到你的Xapian索引的文件系统位置。
import os
HAYSTACK_CONNECTIONS = {
‘default‘: {
‘ENGINE‘: ‘xapian_backend.XapianEngine‘,
‘PATH‘: os.path.join(os.path.dirname(__file__), ‘xapian_index‘),
},
}
```

## 4.处理数据

### 创建索引

如果你想针对某个app例如blog做全文检索，则必须在blog的目录下面建立`search_indexes.py`文件，文件名不能修改

```python
from haystack import indexes
from app01.models import Article

class ArticleIndex(indexes.SearchIndex, indexes.Indexable):
#类名必须为需要检索的Model_name+Index，这里需要检索Article，所以创建ArticleIndex
text = indexes.CharField(document=True, use_template=True)#创建一个text字段
#其它字段
desc = indexes.CharField(model_attr=‘desc‘)
content = indexes.CharField(model_attr=‘content‘)

def get_model(self):#重载get_model方法，必须要有！
return Article

def index_queryset(self, using=None):
return self.get_model().objects.all()
```

为什么要创建索引？索引就像是一本书的目录，可以为读者提供更快速的导航与查找。在这里也是同样的道理，当数据量非常大的时候，若要从这些数据里找出所有的满足搜索条件的几乎是不太可能的，将会给服务器带来极大的负担。所以我们需要为指定的数据添加一个索引（目录），在这里是为Note创建一个索引，索引的实现细节是我们不需要关心的，至于为它的哪些字段创建索引，怎么指定 ,下面开始讲解

每个索引里面必须有且只能有一个字段为 document=True，这代表haystack 和搜索引擎将使用此字段的内容作为索引进行检索(primary field)。其他的字段只是附属的属性，方便调用，并不作为检索数据

```python
注意：如果使用一个字段设置了document=True，则一般约定此字段名为text，这是在ArticleIndex类里面一贯的命名，以防止后台混乱，当然名字你也可以随便改，不过不建议改。
```

另外，我们在`text`字段上提供了`use_template=True`。这允许我们使用一个数据模板（而不是容易出错的级联）来构建文档搜索引擎索引。
你应该在模板目录下建立新的模板`search/indexes/blog/article_text.txt`，并将下面内容放在里面。

```python
#在目录“templates/search/indexes/应用名称/”下创建“模型类名称_text.txt”文件
{{ object.title }}
{{ object.desc }}
{{ object.content }}
```

这个数据模板的作用是对`Note.title`, `Note.user.get_full_name`,`Note.body`这三个字段建立索引，当检索的时候会对这三个字段做全文检索匹配

## 5.设置视图

### 添加`SearchView`到你的`URLconf`

在你的`URLconf`中添加下面一行：

```python
(r‘^search/‘, include(‘haystack.urls‘)),
```

这会拉取Haystack的默认URLconf，它由单独指向`SearchView`实例的URLconf组成。你可以通过传递几个关键参数或者完全重新它来改变这个类的行为。

### 搜索模板

你的搜索模板(默认在`search/search.html`)将可能非常简单。下面的足够让你的搜索运行(你的`template/block`应该会不同)

```python
<!DOCTYPE html>
<html>
<head>
<title></title>
<style>
span.highlighted {
color: red;
}
</style>
</head>
<body>
{% load highlight %}
{% if query %}
<h3>搜索结果如下：</h3>
{% for result in page.object_list %}
{# <a href="/{{ result.object.id }}/">{{ result.object.title }}</a><br/>#}
<a href="/{{ result.object.id }}/">{% highlight result.object.title with query max_length 2%}</a><br/>
<p>{{ result.object.content|safe }}</p>
<p>{% highlight result.content with query %}</p>
{% empty %}
<p>啥也没找到</p>
{% endfor %}

{% if page.has_previous or page.has_next %}
<div>
{% if page.has_previous %}
<a href="?q={{ query }}&amp;page={{ page.previous_page_number }}">{% endif %}&laquo; 上一页
{% if page.has_previous %}</a>{% endif %}
|
{% if page.has_next %}<a href="?q={{ query }}&amp;page={{ page.next_page_number }}">{% endif %}下一页 &raquo;
{% if page.has_next %}</a>{% endif %}
</div>
{% endif %}
{% endif %}
</body>
</html>
```

需要注意的是`page.object_list`实际上是`SearchResult`对象的列表。这些对象返回索引的所有数据。它们可以通过`{{result.object}}`来访问。
所以`{{ result.object.title}}`实际使用的是数据库中Article对象来访问`title`字段的。

### 重建索引

现在你已经配置好了所有的事情，是时候把数据库中的数据放入索引了。Haystack附带的一个命令行管理工具使它变得很容易。

简单的运行`./manage.py rebuild_index`。你会得到有多少模型进行了处理并放进索引的统计。

## 6.使用jieba分词

```python
#建立ChineseAnalyzer.py文件
#保存在haystack的安装文件夹下，路径如“D:python3Libsite-packageshaystackackends”

import jieba
from whoosh.analysis import Tokenizer, Token

class ChineseTokenizer(Tokenizer):
def __call__(self, value, positions=False, chars=False,
keeporiginal=False, removestops=True,
start_pos=0, start_char=0, mode=‘‘, **kwargs):
t = Token(positions, chars, removestops=removestops, mode=mode,
**kwargs)
seglist = jieba.cut(value, cut_all=True)
for w in seglist:
t.original = t.text = w
t.boost = 1.0
if positions:
t.pos = start_pos + value.find(w)
if chars:
t.startchar = start_char + value.find(w)
t.endchar = start_char + value.find(w) + len(w)
yield t

def ChineseAnalyzer():
return ChineseTokenizer()

```

```python
#复制whoosh_backend.py文件，改名为whoosh_cn_backend.py
#注意：复制出来的文件名，末尾会有一个空格，记得要删除这个空格
from .ChineseAnalyzer import ChineseAnalyzer
查找
analyzer=StemmingAnalyzer()
改为
analyzer=ChineseAnalyzer()
```

## 7.在模版中创建搜索栏

```html
<form method=‘get‘ action="/search/" target="_blank">
<input type="text" name="q">
<input type="submit" value="查询">
</form>
```

## 8.其它配置

### 增加更多变量

```python

from haystack.views import SearchView
from .models import *

class MySeachView(SearchView):
def extra_context(self): #重载extra_context来添加额外的context内容
context = super(MySeachView,self).extra_context()
side_list = Topic.objects.filter(kind=‘major‘).order_by(‘add_date‘)[:8]
context[‘side_list‘] = side_list
return context

#路由修改
url(r‘^search/‘, search_views.MySeachView(), name=‘haystack_search‘),
```

### 高亮显示

```python
{% highlight result.summary with query %}
# 这里可以限制最终{{ result.summary }}被高亮处理后的长度
{% highlight result.summary with query max_length 40 %}

#html中
<style>
span.highlighted {
color: red;
}
</style>
```