Redis与Celery基础

Posted 2020-12-18 Francis

目录

• 一 Redis基础
  • 1.1 Redis安装
  • 1.2 Python操作Redis之普通连接
  • 1.3 Python操作Redis之连接池
  • 1.4 Redis操作String
  • 1.5 Redis操作Hash(字典)
  • 1.6 Redis操作List(列表)
  • 1.7 redsi的其他使用
  • 1.8 管道(Redis事务的支持方式)
  • 1.9 Django使用Redis
    • 1.9.1接口缓存实战
• 二 Celery基础
  • 2.1 基本结构,在同一个包下添加任务
  • 2.2 包结构,在不同包下添加任务
    • 2.2.1 执行定时任务

一 Redis基础

1.1 Redis安装

#1 key-value的存储方式，
value有很多数据类型：5大：string(字符串)、list(链表)、set(集合)、zset(sorted set --有序集合)和hash（哈希类型
#2 非关系型内存数据库
#3 与Memcached比较：
	-1 Memcached只支持一种数据类型:字符串
    -2 Memcached不支持持久化（不支持存到硬盘上，只要一断电，数据就没了）
    
# 4 使用Redis有哪些好处？
(1) 速度快，因为数据存在内存中，类似于HashMap，HashMap的优势就是查找和操作的时间复杂度都是O(1)
(2) 支持丰富数据类型，支持string，list，set，sorted set，hash
(3) 支持事务，操作都是原子性，所谓的原子性就是对数据的更改要么全部执行，要么全部不执行
(4) 丰富的特性：可用于缓存，消息，按key设置过期时间，过期后将会自动删除
# 5 单线程，单进程，不存在并发访问的问题（新版本已经不是了）
	-单线程为什么这么快
    	-数据在内存（最重要的）
        -io多路复用技术
        -因为没有进程，线程间的切换
# 6 redis适合的场景
	1 排行榜
    2 网站访问量，文章访问量
    3 缓存数据库（用的最多，就是做缓存）
    4 发布订阅
    5 去重
    6 分布式（简单的消息队列blpop）
    
# 7 安装
	-Redis-x64-3.2.100.msi  安装包
    -redis-desktop-manager-0.9.3.817.exe 等同于navicate
    
    
# 8 使用
	-服务端和客户端
	-安装完，服务端自动启动：redis-server 配置文件.conf
	-redis-cli  ：客户端连接服务端(redis-cli -h 127.0.0.1 -p 6379)

1.2 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‘)

1.3 Python操作Redis之连接池

#redis连接池,redis 的QPS能达到10w
import redis

POOL = redis.ConnectionPool(host=‘127.0.0.1‘, port=6379,max_connections=100)  
# 造一个池子,最多能放100个连接,pool必须是单例的,这里使用导包的方式实现



import redis
#包内的py文件，如果想右键运行，导包的时候不能带点
from t_redis_pool import POOL 
r = redis.Redis(connection_pool=POOL)  # 只要执行这一句话，就是从池中拿出一个连接
ret=r.get(‘name‘)

1.4 Redis操作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 set组合用法(没什么实际作用)
setnx(name, value)
设置值，只有name不存在时，执行设置操作（添加）,如果存在，不会修改
setex(name, value, time)
设置值
参数：
time，过期时间（数字秒 或 timedelta对象）
psetex(name, time_ms, value)
设置值
参数：
time_ms，过期时间（数字毫秒 或 timedelta对象


#mset 依次设置多个减少io操作
conn.mset({‘name1‘:‘11‘,‘name3‘:‘dasfd‘})

ret=conn.mget([‘name1‘,‘name‘,‘name3‘])
ret=conn.getset(‘name1‘, ‘999‘) #取完再设置
ret=conn.getrange(‘name1‘,0,0) # 前闭后闭区间,取一部分防止一次去的太多
ret=conn.setrange(‘name1‘,1,88888)
ret=conn.getbit(‘name1‘,9)


incr ：统计网站访问量，页面访问量，接口访问量
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

1.5 Redis操作Hash(字典)

# hash操作

单个操作
conn.hset(‘hash1‘,‘name‘,‘lqz444‘)  # key不可以重复，
前三个参数分别是:字典名,字典的一个key,字典的一个value
ret=conn.hget(‘hash1‘,‘name‘)  #只能取一个

多个操作
conn.hmset(‘hash2‘,{‘key1‘:‘value1‘,‘key2‘:‘value2‘})
ret=conn.hmget(‘hash1‘,‘name‘,‘name2‘)
ret=conn.hmget(‘hash1‘,[‘name‘,‘name2‘])

其他操作
ret=conn.hgetall(‘hash1‘)  # 尽量少用
ret=conn.hlen(‘hash1‘)
ret=conn.hkeys(‘hash1‘)
ret=conn.hexists(‘hash1‘,‘name1‘)
ret=conn.hdel(‘hash1‘,‘name‘)

重要操作
ret=conn.hincrby(‘hash1‘,‘name‘) 执行一次name的值加一

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

重点总结
hset
hget
hmset
hmget
hincrby
hscan_iter

1.6 Redis操作List(列表)

增
ret=conn.lpush(‘list1‘,1,2,3,4,5) 一次取出从左边放入
ret=conn.rpush(‘list1‘,999)
ret=conn.lpushx(‘list2‘,1) 有才放,没有就不放

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.llen(‘list1‘) 测长度


删
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‘) 

重点block,阻塞,是一个简单的消息队列,可以写一个超时时间
ret=conn.blpop(‘list1‘,timeout=10) 
没有值就一直等着,timeout设置等待时间

ret=conn.ltrim(‘list1‘,1,2) 移除不在区间的值

取
ret=conn.lindex(‘list1‘,0) 取出某个索引位置的值
ret=conn.lrange(‘list1‘,0,2)  # 前闭后闭,切片取

# 自定制分批取列表的数据
#lpush只接受一个个的参数,如果传一个列表需要用*打散

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])


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

使用:            
for item in scan_list(‘test‘,5):
    print(‘---‘)
    print(item)
    
#虽然也是全部取出结果,但是使用了yield生成器大大的节省了内存空间
    
# 重点
lpush
lpop
blpop
lrange
llen

1.7 redsi的其他使用

# 其他操作
conn.delete(‘list1‘)

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

ret=conn.expire(‘hash2‘,2) 设置过期时间

ret=conn.type(‘name3‘) 查看key对应的类型

1.8 管道(Redis事务的支持方式)

一般情况非关系型数据库都不支持事务,Redis通过管道的方式来支持事务

conn = redis.Redis(connection_pool=pool)
pipe = conn.pipeline(transaction=True)

pipe.multi()
pipe.set(‘name‘, ‘alex‘)
pipe.set(‘role‘, ‘sb‘)
pipe.execute()  # 这句话，才真正的去执行

1.9 Django使用Redis

# 方式一（通用方式）
任何框架都可以使用的方式,自己写包导入,一系列操作

# 方式二：django-redis,别人封装好的直接对应Django的缓存系统
	-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,django-redis重写了和cash类一毛一样的方法,在setting中替换了之后,就直接操作Redis了,操作更简便不用关系数据类型
    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‘))

1.9.1接口缓存实战

# 首页轮播图数据缓存到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:
        # 缓存中没有，去数据库拿
        response = super().list(request, *args, **kwargs)
        # 加到缓存
        cache.set(‘banner_list‘,response.data,60*60*24)
        return response 
    #这个response就是一个Response产生的对象可以直接返回

    return Response(data=banner_list)

二 Celery基础

# 1 Celery是一个异步任务框架，执行异步任务，执行延迟任务，执行定时任务
	框架由3部分组成
    broker任务队列(一个消息队列)
    worker任务处理器(去任务队列拿去任务处理,结果放到结果队列中)
    backend结果队列(一个消息队列)

# 2 安装
	-pip install celery

2.1 基本结构,在同一个包下添加任务

#1,得到celery对象  
只写一个py文件，内容如下celery_task.py：
from celery import Celery

broker=‘redis://127.0.0.1:6379/1‘  #Redis的路由协议
backend=‘redis://127.0.0.1:6379/2‘   # 结构存储，执行完的结果存在这

app = Celery(__name__,broker=broker,backend=backend)

@app.task
def addfunc(x,y):
    print(x,y)
    return x+y
#2 cd对对应文件夹下用命令启动celery

	windows平台:
        pip install eventlet
        celery worker -A cele -l info -P eventlet
        
    非windows:
        celery worker -A celery_task -l info

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

2.2 包结构,在不同包下添加任务

#1 新建一个包，叫celery_task
    -celery_task
        -__init__.py
        -celery.py
        -task1.py
        -task2.py
# 2 celery.py 名字必须叫celery
    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 任务写在其他py文件内
	task1.py 
    from .celery import app
    @app.task
    def add(x,y):
        print(x,y)
        return x+y
    
    task2.py
	from .celery import app
    @app.task
    def mutile(x,y):
        print(x,y)
        return x*y
    
# 4 添加任务（异步任务，延迟任务）
    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获取结果同上

2.2.1 执行定时任务

#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