利用Celery实现Django博客PV统计功能详解

发布时间：2024/01/26

我来为你详细讲解“利用Celery实现Django博客PV统计功能详解”的完整攻略。

一、背景介绍

在开发Django博客时，我们经常需要对文章和网站的访问量进行统计，以便更好地了解用户的行为和需求。而Celery是一个常用的异步任务队列，可以方便地实现Django博客的PV统计功能。

二、准备工作

在开始之前，我们需要先安装Celery和Redis：

pip install celery redis

同时，在我们的Django项目中，需要进行如下配置：

1. settings.py

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'

INSTALLED_APPS = [
    ...
    'django_celery_results',
    ...
]

CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

CELERY_BEAT_SCHEDULE = {
    'pv-statistics-task': {
        'task': 'blog.tasks.statistic_pv',
        'schedule': crontab(minute=0, hour='*/1')
    },
}

2. celery.py

import os

from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mysite.settings')

app = Celery('mysite')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

3. tasks.py

from django.core.cache import cache
from django.db import transaction
from django.db.models import F

from blog.models import Post

from celery import shared_task


@shared_task
def statistic_pv():
    cache_keys = cache.keys('views:*')
    view_counts = cache.get_many(cache_keys)
    with transaction.atomic():
        for pk, count in view_counts.items():
            Post.objects.filter(pk=pk).update(views=F('views')+count)
        cache.delete_many(cache_keys)

三、实现步骤

完整的实现步骤如下：

1. 在Django项目中添加缓存

我们需要给Django项目添加一个缓存来统计访问量。这里我们可以使用Django自带的cache模块。

from django.core.cache import cache

def post_detail(request, pk):
    post = get_object_or_404(Post, pk=pk)
    cache_key = f'views:{post.pk}'
    if not cache.get(cache_key):
        post.views += 1
        post.save()
        cache.set(cache_key, 1)
    return render(request, 'blog/post_detail.html', {'post': post})

2. 添加Celery任务

接下来，我们需要使用Celery来定期统计缓存中的访问量。我们可以使用Celery的@shared_task装饰器来定义一个任务，同时将其添加到Celery的调度列表中。

from django.core.cache import cache
from django.db import transaction
from django.db.models import F

from blog.models import Post

from celery import shared_task

@shared_task
def statistic_pv():
    cache_keys = cache.keys('views:*')
    view_counts = cache.get_many(cache_keys)
    with transaction.atomic():
        for pk, count in view_counts.items():
            Post.objects.filter(pk=pk).update(views=F('views')+count)
        cache.delete_many(cache_keys)

3. 定期执行Celery任务

最后，我们需要在Django中添加一个定时任务来周期性地执行Celery任务。这可以通过在CELERY_BEAT_SCHEDULE中添加任务来实现：

CELERY_BEAT_SCHEDULE = {
    'pv-statistics-task': {
        'task': 'blog.tasks.statistic_pv',
        'schedule': crontab(minute=0, hour='*/1')
    },
}

以上代码指定了一个名为pv-statistics-task的任务，可以周期性地（每小时）执行我们定义的statistic_pv任务。

四、实例说明

下面是两条示例说明：

示例一：在Django的视图函数中统计访问量

我们可以在一个Django视图函数中使用缓存来统计访问量。实现过程中，我们需要使用Django缓存模块的cache对象来对访问数进行记录，并使用Post模型来更新记录：

from django.core.cache import cache

def post_detail(request, pk):
    post = get_object_or_404(Post, pk=pk)
    cache_key = f'views:{post.pk}'
    if not cache.get(cache_key):
        post.views += 1
        post.save()
        cache.set(cache_key, 1)
    return render(request, 'blog/post_detail.html', {'post': post})

示例二：使用Celery异步任务统计访问量

我们也可以使用Celery来实现异步的访问量统计。这样做的好处在于，可以避免因为访问量过大影响用户体验。实现过程中，我们需要使用Celery的装饰器@shared_task来定义任务，同时使用Django的事务机制来更新记录：

from django.core.cache import cache
from django.db import transaction
from django.db.models import F

from blog.models import Post

from celery import shared_task

@shared_task
def statistic_pv():
    cache_keys = cache.keys('views:*')
    view_counts = cache.get_many(cache_keys)
    with transaction.atomic():
        for pk, count in view_counts.items():
            Post.objects.filter(pk=pk).update(views=F('views')+count)
        cache.delete_many(cache_keys)

在schedules.py文件中，我们将任务添加到Celery的调度列表中：

from celery.schedules import crontab

from mysite.celery import app

app.conf.beat_schedule = {
    'pv-statistics-task': {
        'task': 'blog.tasks.statistic_pv',
        'schedule': crontab(minute=0, hour='*/1')
    },
}