• Redis

    Redis

    一 简介

    安装

    win版

    最新5.x版本 https://github.com/tporadowski/redis/releases/
    最新3.x版本 https://github.com/microsoftarchive/redis/releases

    二 python操作redis

    安装

    pip install redis

    • 1
    from redis import Redis

conn=Redis()

conn.set('name','xxx')
print(conn.get('name'))

conn.close()

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    三 Redis数据类型操作

    在这里插入图片描述
    Redis 支持的数据类型概述

    Redis 是一个数据结构服务器。 Redis 的核心是提供一系列本机数据类型，可帮助您解决从缓存到队列再到事件处理的各种问题。

    2.1 字符串

    Redis 字符串是最基本的 Redis数据类型，表示字节序列。

    set()

    def set(
    self,
    name: KeyT,
    value: EncodableT,
    ex: Union[ExpiryT, None] = None,  # 过期时间（秒）
    px: Union[ExpiryT, None] = None,  # 过期时间（毫秒）
    nx: bool = False,  # 如果设置为True，则只有name不存在时，当前set操作才执行,值存在，就修改不了，执行没效果。
    xx: bool = False,  # 如果设置为True，则只有name存在时，当前set操作才执行，值存在才能修改，值不存在，不会设置新值。
    keepttl: bool = False,  # 如果为True，保留与该键关联的生存时间。
    get: bool = False,  # 如果为True，将键name处的值设置为value并返回旧值存储在key中，如果key不存在则为None。
    exat: Union[AbsExpiryT, None] = None,  # 设置指定的Unix时间，该键将在该时间到期，单位为秒。
    pxat: Union[AbsExpiryT, None] = None,  # 设置指定的Unix时间，该键将在该时间到期，单位为毫秒。

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    import redis

conn = redis.Redis()
conn.set('name', 'xuxiaoxu')
conn.set('age', 18)

conn.close()

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    setnx()

    # 相当于set(..., nx=True)
def setnx(self, name: KeyT, value: EncodableT) -> ResponseT:

    • 1
    • 2
    conn.setnx('name','xxx')  # 存在不修改，执行没效果
conn.setnx('hobby','pingpang')

    • 1
    • 2

    psetex()

    # 本质就是 set(..., px)设置时间(毫秒)
def psetex(self, name: KeyT, time_ms: ExpiryT, value: EncodableT):

    • 1
    • 2
    conn.psetex('name', 3000, 'jasper')

    • 1

    mset()

    # 批量设置
def mset(self, mapping: Mapping[AnyKeyT, EncodableT]) -> ResponseT:

    • 1
    • 2
    conn.mset({'name': 'xuxiaoxu', 'age': 20, 'xxx': 'yyy'})

    • 1

    get()

    def get(self, name: KeyT) -> ResponseT:

    • 1
    # 获取值，取到是bytes格式
# 可以在实例化redis对象时指定参数decode_responses=True
res = conn.get('name')
print(res)  # b'xuxiaoxu'

    • 1
    • 2
    • 3
    • 4

    mget()

    # 批量获取
def mget(self, keys: KeysT, *args: EncodableT) -> ResponseT:

    • 1
    • 2
    res = conn.mget('name', 'age')
print(res)  # [b'xuxiaoxu', b'20']

    • 1
    • 2

    getset()

    # 先获取在设置
def getset(self, name: KeyT, value: EncodableT) -> ResponseT:

    • 1
    • 2
    res = conn.getset('age', 30)
print(res)  # b'20'

    • 1
    • 2

    getrange()

    # 取的是字节，左闭右闭区间。
def getrange(self, key: KeyT, start: int, end: int) -> ResponseT:

    • 1
    • 2
    res = conn.getrange('name',0,3)
print(res)  # b'xuxi'

    • 1
    • 2

    setrange()

    '''
从offset开始重新设置值，如果``offset``加上``value``的长度超过原始值的长
度，新值将比以前大。如果``offset``超过了原始值的长度,则用null填充。
'''
def setrange(self, name: KeyT, offset: int, value: EncodableT) -> ResponseT:

    • 1
    • 2
    • 3
    • 4
    • 5
    res = conn.setrange('name',10, 'ishandsome')
print(res)  # 20

    • 1
    • 2

    在这里插入图片描述
    setbit()

    # 设置或清除存储在键的字符串值中的偏移位。
# 根据值设置或清除位，该值可以是 0 或 1.
def setbit(self, name: KeyT, offset: int, value: int) -> ResponseT:

    • 1
    • 2
    • 3

    bitcount()

    # 返回` ' key ' `的值中设置的位数。可选``start``和``end``参数表示要考虑哪些字节
def bitcount(
    self,
    key: KeyT,
    start: Union[int, None] = None,
    end: Union[int, None] = None,
    mode: Optional[str] = None,
) -> ResponseT:

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    res = conn.bitcount('hobby', 0, 6)
print(res)  # 28

    • 1
    • 2

    strlen()

    # 返回` ' name ' `的值中存储的字节数
def strlen(self, name: KeyT) -> ResponseT:

    • 1
    • 2
    print(conn.strlen('hobby'))  # 8

    • 1

    incr()

    def incr(self, name: _Key, amount: int = ...) -> int: ...

    • 1
    res = conn.incr('age', amount=2)
print(res)

    • 1
    • 2

    incrbyfloat()

    def incrbyfloat(self, name: KeyT, amount: float = 1.0) -> ResponseT:

    • 1

    decr()

    def decr(self, name, amount: int = ...) -> int: ...

    • 1
    print(conn.decr('age'))

    • 1

    append()
    将字符串’ ’ value ’ ‘附加到' key '的值上。如果“键”,不存在value创建它。返回值’ ’ key ’ '的新长度。

    def append(self, key: KeyT, value: EncodableT) -> ResponseT:

    • 1
    print(conn.append('name', 'hahaha'))  # 14

    • 1

    2.2 hash

    hset()

    '''
在hash `name`中设置`key`为`value`，
``mapping``接受一个由键值对组成的字典
添加到hash ``name``中。
``items``接受一个键值对列表作为参数
添加到hash ``name``中。
返回添加的字段数。
'''
def hset(
     self,
     name: str,
     key: Optional[str] = None,
     value: Optional[str] = None,
     mapping: Optional[dict] = None,
     items: Optional[list] = None,
 ) -> Union[Awaitable[int], int]:

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    res = conn.hset('info', 'name', 'xuxiaoxu', mapping={'age': 18, 'xx': 'yy'})
print(res)  # 3

    • 1
    • 2

    hmset()
    将key设置为对应哈希name中的值，mapping字典中的键和值。

    def hmset(self, name: str, mapping: dict) -> Union[Awaitable[str], str]:

    • 1
    res = conn.hmset('info',{'xx':'xx', 'yy':'yy'})
print(res)
#  DeprecationWarning: Redis.hmset() is deprecated. Use Redis.hset() instead.
#  res = conn.hmset('info',{'xx':'xx', 'yy':'yy'})
# 已被弃用,使用hset()即可。

    • 1
    • 2
    • 3
    • 4
    • 5

    hget()

    返回哈希值key中的值name

    def hget(
    self, name: str, key: str
) -> Union[Awaitable[Optional[str]], Optional[str]]:

    • 1
    • 2
    • 3
    res = conn.hget('info', 'xx')
print(res)  # b'xx'

    • 1
    • 2

    hmget()

    返回与keys顺序相同的值的列表。

    def hmget(self, name: str, keys: List, *args: List) -> Union[Awaitable[List], List]:

    • 1
    res = conn.hmget('info', ('xx', 'yy'))
print(res)  # [b'xx', b'yy']

    • 1
    • 2

    hgetall()

    返回一个Python字典，由散列的名称/值对组成

    def hgetall(self, name: str) -> Union[Awaitable[dict], dict]:

    • 1
    res = conn.hgetall('info')
print(res)  # {b'name': b'xuxiaoxu', b'age': b'18', b'xx': b'xx', b'yy': b'yy'}

    • 1
    • 2

    hlen()

    返回哈希值name中的元素个数

    def hlen(self, name: str) -> Union[Awaitable[int], int]:

    • 1
    res = conn.hlen('info')
print(res)  # 4

    • 1
    • 2

    hkeys()

    返回hash name 中的键列表

    def hkeys(self, name: str) -> Union[Awaitable[List], List]:

    • 1
    res = conn.hkeys('info')
print(res)  # [b'name', b'age', b'xx', b'yy']

    • 1
    • 2

    hvals()

    返回hash name中的值列表

    def hvals(self, name: str) -> Union[Awaitable[List], List]:

    • 1
    res = conn.hvals('info')
print(res)  # [b'xuxiaoxu', b'18', b'xx', b'yy']

    • 1
    • 2

    hexists()

    返回一个布尔值，表示key是否存在于哈希name中。

    def hexists(self, name: str, key: str) -> Union[Awaitable[bool], bool]:

    • 1
    res = conn.hexists('info','xx')
print(res)  # True

    • 1
    • 2

    hdel()

    从hash name中删除keys

    def hdel(self, name: str, *keys: List) -> Union[Awaitable[int], int]:

    • 1
    res = conn.hdel('info','xx', 'yy')
print(res)  # 2

    • 1
    • 2

    hincrby()

    将hash name中的key的值增加amount

    def hincrby(
    self, name: str, key: str, amount: int = 1
) -> Union[Awaitable[int], int]:

    • 1
    • 2
    • 3
    conn.hset('ingo','age',18)
res = conn.hincrby('info','age',2)
print(res)  # 20

    • 1
    • 2
    • 3

    hincrbyfloat()

    将hash name中的key的值增加amount

    def hincrby(
    self, name: str, key: str, amount: int = 1
) -> Union[Awaitable[int], int]:

    • 1
    • 2
    • 3

    hscan()

    在哈希表中递增地返回键值对切片，同样返回一个游标。
    表示扫描位置。
    match允许按模式过滤键
    count允许提示最小返回次数

    def hscan(
    self,
    name: KeyT,
    cursor: int = 0,
    match: Union[PatternT, None] = None,
    count: Union[int, None] = None,
) -> ResponseT:

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    # count不准确，有点上下浮动。
res = conn.hscan('test', cursor=0, count=7)
print(res)
print(len(res[1]))
print(res[0])
res = conn.hscan('test', cursor=res[0], count=7)
print(res)
print(len(res[1]))
print(res[0])

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    在这里插入图片描述
    hscan_iter()

    使用HSCAN命令创建一个迭代器，这样客户端就不需要迭代，需要记住光标的位置。
    match允许按模式过滤键
    count允许提示最小返回次数

    def hscan_iter(
     self,
     name: str,
     match: Union[PatternT, None] = None,
     count: Union[int, None] = None,
 ) -> Iterator:
    cursor = "0"
    while cursor != 0:
        cursor, data = self.hscan(name, cursor=cursor, match=match, count=count)
        yield from data.items()

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    res = conn.hscan_iter('test')
print(res)  # 生成器 
 

for i in res:
    print(i)
conn.close()

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    2.3 列表操作

    lpush()

    将values推到列表的头部name

    def lpush(self, name: str, *values: FieldT) -> Union[Awaitable[int], int]:

    • 1
    res = conn.lpush('redis_list', 1, 2, 3, 4, 5)
print(res)  # 5

    • 1
    • 2

    lpushx()

    如果name存在，将value推到name的头部

    def lpushx(self, name: str, *values: FieldT) -> Union[Awaitable[int], int]:

    • 1
    res = conn.lpushx('redis_list', 6, 7, 8)
print(res)  # 8

    • 1
    • 2

    rpush()和rpushx()

    与上面两个相反，推到列表的尾部。

    llen()

    返回列表的长度name

    def llen(self, name: str) -> Union[Awaitable[int], int]:

    • 1
    res = conn.llen('redis_list')
print(res)  # 8

    • 1
    • 2

    linsert()

    在列表中插入元素，该元素存储在基准值基准值之前或之后。
    如果key不存在，则认为它是一个空列表，不会执行任何操作。
    如果key存在但没有保存列表值，则返回错误。

    def linsert(
     self, name: str, where: str, refvalue: str, value: str
 ) -> Union[Awaitable[int], int]:

    • 1
    • 2
    • 3
    res = conn.linsert('redis_list', 'before', '3', 'new')
print(res)  # 9

    • 1
    • 2

    lset()

    将list name的index元素设置为value

    def lset(self, name: str, index: int, value: str) -> Union[Awaitable[str], str]:

    • 1
    res = conn.lset('redis_list', 0, 'lset')
print(res)  # True

    • 1
    • 2

    lrem()

    删除第一次count等于value的元素从存储在name中的列表中。

    def lrem(self, name: str, count: int, value: str) -> Union[Awaitable[int], int]:

    • 1
    res = conn.lrem('redis_list', 1, '5')
print(res)  # 1

    • 1
    • 2

    lpop()

    删除并返回列表name的第一个元素。

    默认情况下，该命令从开头弹出一个元素从列表中。当提供可选的count参数时，响应将由多数量的元素组成，这取决于列表的长度。

    从 Redis 版本 6.2.0 开始：添加了参数count。

    def lpop(self, name: str, count: Optional[int] = None) -> Union[str, List, None]:

    • 1
    res = conn.lpop('redis_list')
print(res)  # b'lset'

    • 1
    • 2

    lindex()

    从列表name中返回位置index
    支持负索引，从列表末尾开始

    def lindex(
     self, name: str, index: int
 ) -> Union[Awaitable[Optional[str]], Optional[str]]:

    • 1
    • 2
    • 3
    res = conn.lindex('redis_list', 0)
res1 = conn.lindex('redis_list', -1)

    • 1
    • 2

    lrange()

    返回列表name之间的一个切片
    定位start和end
    start和end也可以是负数
    Python切片表示法

    def lrange(self, name: str, start: int, end: int) -> Union[Awaitable[list], list]:

    • 1
    res = conn.lrange('redis_list',0,-1)
print(res)

    • 1
    • 2

    ltrim()

    修剪列表name，删除所有不在切片内的
    在start和end之间

    start和end也可以是负数

    Python切片表示法

    def ltrim(self, name: str, start: int, end: int) -> Union[Awaitable[str], str]:

    • 1
    res = conn.ltrim('redis_list',1,-2)

    • 1

    rpoplpush()

    从src列表中取出一个值RPOP，然后在dst列表上LPUSH。并返回rpop的值。

    def rpoplpush(self, src: str, dst: str) -> Union[Awaitable[str], str]:

    • 1
    conn.lpush('new_list', '1')
res = conn.rpoplpush('redis_list', 'new_list')
print(res)  # b'new'

    • 1
    • 2
    • 3

    blpop()

    从第一个非空列表中取出一个值

    在 keys 列表中命名。

    如果keys中的列表都没有值可以LPOP，则阻塞timeout秒，或者直到一个值被推送给一个值名单的。

    如果timeout为0，则无限期阻塞。

    def blpop(
     self, keys: List, timeout: Optional[int] = 0
 ) -> Union[Awaitable[list], list]:

    • 1
    • 2
    • 3
    res = conn.blpop('redis_list')
print(res)

    • 1
    • 2

    brpoplpush()

    BRPOPLPUSH是RPOPLPUSH 的阻塞变体。 当包含元素时，此命令的行为与RPOPLPUSH 完全相同。 当在MULTI/EXEC块中使用时，此命令的行为与RPOPLPUSH 完全相同。 当为空时，Redis 将阻止连接，直到另一个客户端 推到它或直到到达。 零点可用于无限期阻止。

    def brpoplpush(
     self, src: str, dst: str, timeout: Optional[int] = 0
 ) -> Union[Awaitable[Optional[str]], Optional[str]]:

    • 1
    • 2
    • 3
    conn.lpush('redis_list', 1, 2, 3, 4)
res = conn.brpoplpush('redis_list', 'new_list')
print(res)

    • 1
    • 2
    • 3

    2.4 其他操作

    五大数据类型都支持

    delete()

    删除一个或多个由names指定的键。

    def delete(self, *names: KeyT) -> ResponseT:

    • 1
    res = conn.delete('test')
print(res)  # 1

    • 1
    • 2

    exists()

    返回存在的names的数量

    def exists(self, *names: KeyT) -> ResponseT:

    • 1
    res = conn.exists('new_list', 'redis_list', 'xxx')
print(res)  # 2

res = conn.exists('xxx')
print(res)  # 0

    • 1
    • 2
    • 3
    • 4
    • 5

    keys()

    返回一个匹配pattern的键列表

    def keys(self, pattern: PatternT = "*", **kwargs) -> ResponseT:

    • 1
    res = conn.keys('*list*')
print(res)  # [b'redis_list', b'new_list']

    • 1
    • 2

    expire()

    在键name上为time 秒设置一个过期标志选项。time可以用整数或Python的timedelta对象表示。

    def expire(
     self,
     name: KeyT,
     time: ExpiryT,
     nx: bool = False,
     xx: bool = False,
     gt: bool = False,
     lt: bool = False,
 ) -> ResponseT:

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    

    • 1
  • 相关阅读:
    Django与Ajax
    Linux Shell入门常用命令使用
    校园网站设计,校园门户网站,学校官网,python计算机毕业设计
    8位ADC模板实例
    【数据集】水文专业常用数据集整理
    docker（二）容器的启动，停止，删除
    java计算机毕业设计vue架构云餐厅美食订餐系统源码+mysql数据库+系统+lw文档+部署
    torch.mean()、tensor.mean() 用法及维度的理解(二维、三维)
    基于fast-Unet的补强胶胶体在线识别分割技术
    RT-Thread GTC 2022 全球技术大会参与指南
  • 原文地址：https://blog.csdn.net/weixin_68531269/article/details/127871190