day95-Redis操作、管道、celery的使用

文章目录

• 昨日回顾
• 今日内容
• • 1 Python操作Redis之普通连接
  • 2 Python操作Redis之连接池
  • 3 操作之String操作
  • 4 操作之Hash操作
  • 5 操作之List操作
  • 5 redsi的其他使用
  • 6 管道
  • 7 Django中使用redis
  • 8 接口缓存
  • 9 celery的使用
  • • 9.1 基本结构
    • 9.2 包结构
    • 9.3 执行定时任务
• 补充
• 作业

昨日回顾

# 1 频率限制类 （继承SimpleRateThrottle，scop，重写getcachekey方法：方法返回什么，就以什么作为key限制，setting中配置）
# 2 手机号验证码登录，手机号和验证码
# 3 注册接口
# 4 redis：安装和启动   
	-5大数据类型：str：接口缓存，list：分布式，hash：缓存，set：去重，zset：排行榜
	-redis-server 以某个配置文件启动  启动（windows上在服务中右键启动也可以）
    -redis-cli -h  -p

今日内容

1 Python操作Redis之普通连接

# 1 pip3 install redis
# 简单使用
from redis import Redis
# conn=Redis()
#连接对象
conn=Redis(host='127.0.0.1', port=6379)
ret=conn.get('name')
print(ret)

2 Python操作Redis之连接池

###### t_redis_pool.py
#redis连接池
import redis
#pool必须是单例的
POOL = redis.ConnectionPool(host='127.0.0.1', port=6379,max_connections=100)  # 造一个池子,最多能放100个连接

#######t_redis_conn.py
#redis连接池
# import redis
# #pool必须是单例的
# pool = redis.ConnectionPool(host='127.0.0.1', port=6379,max_connections=100)  # 造一个池子,最多能放100个连接
import redis
#包内的py文件，如果想右键运行，导包的时候不能带点
from t_redis_pool import POOL  # pycharm提示的错
r = redis.Redis(connection_pool=POOL)  # 只要执行这一句话，就是从池中拿出一个连接
ret=r.get('name')
print(ret)

3 操作之String操作

#####字符串操作
####1 set的用法
# conn.set('height',180) #基本使用

# conn.set('height','190',nx=True)
# conn.set('height','190',xx=True)
# conn.set('height1','190',xx=True)
'''
ex，过期时间（秒）
     px，过期时间（毫秒）
     nx，如果设置为True，则只有name不存在时，当前set操作才执行,值存在，就修改不了，执行没效果
     xx，如果设置为True，则只有name存在时，当前set操作才执行，值存在才能修改，值不存在，不会设置新值
 
'''

### 2
# setnx(name, value)
#
# 设置值，只有name不存在时，执行设置操作（添加）,如果存在，不会修改
# setex(name, value, time)
# # 设置值
# # 参数：
#     # time，过期时间（数字秒 或 timedelta对象）
# psetex(name, time_ms, value)
# # 设置值
# # 参数：
#     # time_ms，过期时间（数字毫秒 或 timedelta对象


# mset
# conn.mset({'name1':'11','name3':'dasfd'})

# ret=conn.mget(['name1','name','name3'])
# print(ret)

# ret=conn.getset('name1', '999')
# print(ret)

# ret=conn.getrange('name1',0,0) # 前闭后闭区间
# print(ret)

# conn.setrange('name1',1,88888)

# ret=conn.getbit('name1',9)
# print(ret)

#incr ：统计网站访问量，页面访问量，接口访问量
# conn.incr('name1')  # 只要一执行，数字加1
# conn.incr('name1')  # 只要一执行，数字加1
# conn.incr('name1')  # 只要一执行，数字加1
# conn.incr('name1')  # 只要一执行，数字加1

#decr
# conn.incr('name1',-2)
# conn.decr('name1',3)

# conn.append('name1','oo')
# conn.incr('name1')

##重点：
#set :很多参数
#get
#mset
#mget
#incr
#decr
#append

4 操作之Hash操作

# hash操作
# conn.hset('hash1','name','lqz')
# conn.hset('hash1','name2','lqz')
# conn.hset('hash1','name','lqz444')  # key不可以重复，

# ret=conn.hget('hash1','name')  #只能取一个
# print(ret)


# conn.hmset('hash2',{'key1':'value1','key2':'value2'})
# ret=conn.hmget('hash1','name','name2')
# ret=conn.hmget('hash1',['name','name2'])
# print(ret)

# ret=conn.hgetall('hash1')  # 尽量少用
# print(ret)

# ret=conn.hlen('hash1')
# ret=conn.hkeys('hash1')
# ret=conn.hexists('hash1','name1')
# ret=conn.hdel('hash1','name')

# conn.hset('hash1','name',12)
# ret=conn.hincrby('hash1','name')
#
# print(ret)


# 以后想取出hash类型内所有的数据，不建议用hgetall，建议用hscan_iter
# 一次性先取一部分回来（假设有1w条，先取回100条，把这100条做成了生成器）
# ret=conn.hscan_iter('hash1')
# print(ret)
# for i in ret:
#     print(i)

##重点掌握
# hset
# hget
#hmset
#hmget
# hincrby
# 区分hgetall和hscan_iter

5 操作之List操作

### 列表操作
# ret=conn.lpush('list1',1,2,3,4,5)
# ret=conn.rpush('list1',999)
# ret=conn.lpushx('list2',1)

# ret=conn.lpushx('list1',888)  # 必须有这个key才能放
# ret=conn.rpushx('list1',666)  # 我们猜，返回总长度
# ret=conn.llen('list1')

# ret=conn.linsert('list1','before','3','77777777')
# ret=conn.linsert('list1','after','3','66666666')



# ret=conn.lset('list1',3,'22222')  #从0开始计数
# ret=conn.lset('list1',0,'11111')

# ret=conn.lrem('list1',2,'5')  # 从前往后删除两个5
# ret=conn.lrem('list1',-1,'5') # 从后往前删除1个5
# ret=conn.lrem('list1',0,'5')   # 删除所有5

# ret=conn.lpop('list1')
# ret=conn.rpop('list1')

# ret=conn.lindex('list1',0)

# ret=conn.lrange('list1',0,2)  # 前闭后闭

# ret=conn.ltrim('list1',1,2)

# 重点block,阻塞,可以写一个超时时间
# ret=conn.blpop('list1',timeout=10)
# print(ret)


# 自定制分批取列表的数据
# conn.lpush('test',*[1,2,3,4,45,5,6,7,7,8,43,5,6,768,89,9,65,4,23,54,6757,8,68])
# conn.flushall()
def scan_list(name,count=2):
    index=0
    while True:
        data_list=conn.lrange(name,index,count+index-1)
        if not data_list:
            return
        index+=count
        for item in data_list:
            yield item
# print(conn.lrange('test',0,100))
for item in scan_list('test',5):
    print('---')
    print(item)

    
# 重点
lpush
lpop
blpop
lrange
llen

5 redsi的其他使用

# 其他操作
# conn.delete('list1')
# ret=conn.delete('hash1')

# ret=conn.exists('hash2')
# ret=conn.keys('cache*')  #查询以cache开头的所有key

# ret=conn.expire('hash2',2)

# ret=conn.type('name3')
# ret=conn.type('test')
# ret=conn.type('test')
print(ret)

6 管道

# redis支持事务
# 管道。实现事务
# import redis
# pool = redis.ConnectionPool(host='127.0.0.1', port=6379)
# 
# conn = redis.Redis(connection_pool=pool)
# 
# # pipe = r.pipeline(transaction=False)
# pipe = conn.pipeline(transaction=True)
# pipe.multi()
# pipe.set('name', 'alex')
# 
# pipe.set('role', 'sb')
# 
# pipe.execute()  # 这句话，才真正的去执行

7 Django中使用redis

# 方式一（通用方式）
# 方式二：django-redis
	-pip install django-redis
    -setting中配置
    	CACHES = {
                "default": {
                    "BACKEND": "django_redis.cache.RedisCache",
                    "LOCATION": "redis://127.0.0.1:6379",
                    "OPTIONS": {
                        "CLIENT_CLASS": "django_redis.client.DefaultClient",
                        "CONNECTION_POOL_KWARGS": {"max_connections": 100}
                        # "PASSWORD": "123",
                    }
                }
            }
   	-使用两种方式
    	1 使用cache
        from django.core.cache import cache
        cache.set('name',user) 
        2 直接使用conn对象
        from django_redis import get_redis_connection
        conn = get_redis_connection('default')
        print(conn.hgetall('xxx'))

8 接口缓存

# 首页轮播图数据缓存到redis中
def list(self, request, *args, **kwargs):

    # response=super().list(request, *args, **kwargs)
    # 把data的数据加缓存
    # 1 先去缓存拿数据
    banner_list=cache.get('banner_list')
    if not banner_list:
        print('走数据库了')
        # 缓存中没有，去数据库拿
        response = super().list(request, *args, **kwargs)
        # 加到缓存
        cache.set('banner_list',response.data,60*60*24)
        return response

    return Response(data=banner_list)

9 celery的使用

# 1 异步任务框架，执行异步任务，执行延迟任务，执行定时任务
# 2 Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.
# 3 使用
	-pip install celery
# 4 两种结构

9.1 基本结构

#1  只写一个py文件，内容如下celery_task.py：
from celery import Celery
broker='redis://127.0.0.1:6379/1'  #broker任务队列
backend='redis://127.0.0.1:6379/2'   # 结构存储，执行完的结果存在这
app=Celery(__name__,broker=broker,backend=backend)
#添加任务(使用这个装饰器装饰，@app.task)
@app.task
def add(x,y):
    print(x,y)
    return x+y
# 2启动worker
        # 用命令来执行
        # 非windows
        # 命令：celery worker -A celery_task -l info
        # windows：
        # pip3 install eventlet
        # celery worker -A celery_task -l info -P eventlet
        
# 3 添加任务
	from celery_task import add
    # add(3,4)  # 直接执行，不会被添加到broker中
    ret=add.delay(5,4)  #想broker中添加一个任务
    print(ret)
# 4 查看任务执行结果
	from celery_task import app
    from celery.result import AsyncResult
    id = '3e397fd7-e0c1-4c5c-999c-2655a96793bb'
    if __name__ == '__main__':
        async = AsyncResult(id=id, app=app)
        if async.successful():
            result = async.get()
            print(result)
        elif async.failed():
            print('任务失败')
        elif async.status == 'PENDING':
            print('任务等待中被执行')
        elif async.status == 'RETRY':
            print('任务异常后正在重试')
        elif async.status == 'STARTED':
            print('任务已经开始被执行')

9.2 包结构

#1 新建一个包，叫celery_task
    -celery_task
        -__init__.py
        -celery.py
        -task1.py
        -task2.py
# 2 celery.py
    from celery import Celery
    broker='redis://127.0.0.1:6379/1'  #broker任务队列
    backend='redis://127.0.0.1:6379/2'# 结构存储，执行完的结果存在这
    app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.task1','celery_task.task2'])
    
# 3 task1.py
    from .celery import app
    @app.task
    def add(x,y):
        print(x,y)
        return x+y
# 4 task2.py
	from .celery import app
    @app.task
    def mutile(x,y):
        print(x,y)
        return x*y
# 5 添加任务（异步任务，延迟任务）
    from celery_task.task1 import add
    from celery_task.task2 import mutile
    #  提交异步
    ret=add.delay(6,7)
    print(ret)  # 2d4ad592-9548-4c7c-8df4-7f8583e8a1b1
    
    # 提交延迟任务
    from datetime import datetime, timedelta
    # 需要utc时间
    eta=datetime.utcnow() + timedelta(seconds=10)
    ret=add.apply_async(args=(240, 50), eta=eta)
    print(ret)

# 6获取结果同上

9.3 执行定时任务

#1 celery.py

    from celery import Celery
    broker='redis://127.0.0.1:6379/1'  #broker任务队列
    backend='redis://127.0.0.1:6379/2'   # 结构存储，执行完的结果存在这
    app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.task1','celery_task.task2'])
    # 执行定时任务
    # 时区
    app.conf.timezone = 'Asia/Shanghai'
    # 是否使用UTC
    app.conf.enable_utc = False
    # 任务的定时配置
    from datetime import timedelta
    from celery.schedules import crontab
    app.conf.beat_schedule = {
        'add-task': {
            'task': 'celery_task.task1.add',
            # 'schedule': timedelta(seconds=3),
            'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
            'args': (300, 150),
        }
    }
    
# 2 启动worker，启动beat
	-celery worker -A celery_task -l info -P eventlet
	-celery beat -A celery_task -l info

补充

## 1 同步调用和异步调用

## 2 缓存雪崩，缓存击穿，缓存穿透

作业

## 1 整理redis的string，list，hash类型的使用，上课讲的方法都敲一下

## 2 使用celery，执行一个异步任务，把传入的字符串，存入文件

## 3 使用celery，执行一个延时任务，延时5s，把传入的字符串存入文件

## 4 执行一个定时任务，每隔3s钟，从log.log文件中取出最后一行数据，存入log_new.log中和存到数据库中

## 5 使用celery，每隔5s，更新一下banner的缓存

## 6 (选做)分析课程的表关系，根据路飞，尝试写出课程的表及关联表