duongnt.bk

Đây là bài đầu tiên trong loạt 3 bài về distributed caching sử dụng Redis. Xin xem trang này để lấy đường link tới các phần còn lại.

Có lẽ redis-py là package kết nối Redis phổ biến nhất cho Python. Tuy nhiên với redis-py, việc gửi lệnh một cách non-blocking là tương đối phức tạp. Chính vì thế mà aioredis đã được phát triển. Package này cho phép ta sử dụng asyncio để kết nối tới Redis. Từ bản 2.0, API của aioredis đã được viết lại cho giống với API của redis-py. Ta có thể cài package này bằng lệnh sau.

pip install aioredis

Thiết lập server Redis để test

Ta cần một server Redis để chạy code ví dụ trong bài này. Nhưng may mắn là ta có thể chạy Redis trong container của Docker.

docker run -d -p 6379:6379 --name local-redis -d redis

Lệnh trên sẽ tải về image này và khởi tạo một container với tên là local-redis. Sau đó ta có thể kết nối tới server Redis đó tại URL sau.

redis://localhost:6379

Ta cũng có thể truyền vào đường dẫn tới file cấu hình nếu muốn.

docker run -d -p 6379:6379 --name local-redis -d redis redis-server <path to redis.conf file>

Kết nối tới Redis

Cách kết nối tới Redis đơn giản nhất là tạo một client và và tái sử dụng kết nối từ một connection pool.

redis = aioredis.from_url('redis://localhost:6379')

Redis lưu dữ liệu dưới dạng bytes. Nhưng nếu ta chỉ muốn lưu dữ liệu string với encoding đã biết trước thì ta có thể đặt thiết lập để client tự động decode tất cả giá trị trả về từ Redis.

# Tự động decode tất cả giá trị trả về thành string, sử dụng utf-8.
redis = aioredis.from_url(
    'redis://localhost:6379',
    encoding='utf-8',
    decode_responses=True)

Và nếu server Redis có mật khẩu thì ta cần nhập cả mật khẩu khi tạo client.

redis = aioredis.from_url(
    'redis://localhost:6379',
    username=<tài khoản>,
    password=<mật khẩu>)

Ngoài ra, aioredis còn cung cấp tùy chọn để thiết lập SSL certificate, cổng, thiết lập cho socket,… Các bạn có thể tham khảo danh sách tại đây, chỉ cần tìm từ khóa __slot__.

Ghi dữ liệu vào Redis bằng aioredis

Redis có nhiều kiểu dữ liệu khác nhau, và với từng kiểu, aioredis cung cấp những hàm tương ứng để ghi/đọc giá trị. Tất cả các hàm đó đều được gọi từ client redis mà ta vừa khởi tạo ở trên. Dưới đây là một số kiểu dữ liệu thường dùng nhất.

  • String
  • List
  • Hash
  • Set
  • SortedSet

Ghi kiểu string vào Redis

Đây là trường hợp đơn giản nhất, ta dùng hàm set.

await redis.set('string_1', 'Example')
await redis.set('string_2', 'Another Example')

Ghi kiểu list vào Redis

Để thêm một giá trị vào đầu list, ta dùng hàm lpush. Nếu key ta sử dụng không tồn tại thì aioredis sẽ tạo một list mới.

await redis.lpush('list_1', 'one')

Ta cũng có thể thêm nhiều giá trị một lúc.

await redis.lpush('list_2', 'two', 'three', 'four') # List sẽ chứa ['four', 'three', 'two']

Tương tự thế, ta có thể thêm một hoặc nhiều giá trị vào cuối list bằng hàm rpush. Lúc này nếu key ta sử dụng không tồn tại thì aioredis cũng sẽ tạo một list mới.

await redis.rpush('list_3', 'one')
await redis.rpush('list_4', 'two', 'four', 'six') # List sẽ chứa ['two', 'four', 'six']

Để thêm một giá trị vào trước hoặc sau một giá trị khác thì ta dùng hàm linsert. Nếu key ta sử dụng không tồn tại hoặc ta thêm giá trị vào trước/sau một giá trị không tồn tại trong list thì lệnh của ta sẽ là noop. Còn nếu giá trị đó lặp lại nhiều lần trong list thì giá trị mới sẽ được thêm vào trước/sau lần xuất hiện đầu tiên của giá trị đó.

redis.linsert('<cache key>', <BEFORE or AFTER>, '<existing value>', '<new value>')

Ví dụ.

await redis.linsert('list_4', 'BEFORE', 'four', 'three') # List sẽ chứa ['two', 'three', 'four', 'six']
await redis.linsert('list_4', 'AFTER', 'four', 'five') # List sẽ chứa ['two', 'three', 'four', 'five', 'six']

Ghi kiểu hash vào Redis

Mặc dù Redis có cả lệnh HSET và HMSET, lệnh hmset trong aioredis đã lỗi thời. Vì vậy, ta sẽ dùng lệnh hset cho cả trường hợp thêm một trường và trường hợp thêm nhiều trường vào hash. Trong cả 2 trường hợp, nếu key ta sử dụng không tồn tại thì một hash mới sẽ được khởi tạo.

Để thêm một trường vào hash.

redis.hset('<cache key>', '<field>', <value>)

Và dưới đây là ví dụ thêm nhiều trường vào hash. Có thể thấy là ta phải khởi tạo một dictionary để lưu các cặp giá trị trong hash. Dictionary này được truyền vào biến mapping. Ta không cần phải truyền tên trường trực tiếp vào hàm hset nữa.

redis.hset('<cache key>', mapping=<dictionary>)

Ví dụ.

await redis.hset('hash_1', 'first_name', 'John')

values = {
    'last_name': 'Doe',
    'age': 100
}
await redis.hset('hash_1', mapping=values)

Hash sẽ chứa các giá trị.

1) "first_name"
2) "John"
3) "last_name"
4) "Doe"
5) "age"
6) "100"

Ghi kiểu set vào Redis

Vì các giá trị trong set là không có thứ tự nên ta chỉ cần một hàm sadd để ghi giá trị mới vào set. Nếu key ta dùng không tồn tại thì một set mới sẽ được khởi tạo.

redis.sadd('<cache key>', <values>)

Ví dụ.

await redis.sadd('set_1', 'one', 'two', 'three', 'four', 'five')

Ghi kiểu sorted set vào Redis

Lúc này ta cần dùng hàm zadd. Vì các giá trị trong sorted set có thứ tự nên khi thêm một giá trị mới, ta cần truyền vào cả giá trị và điểm xếp hạng. Các cặp giá trị và điểm xếp hàng này được lưu trong một dictionary. Nếu key ta sử dụng không tồn tại thì một sorted set mới sẽ được khởi tạo.

redis.zadd('<cache key>', <dictionary>)

Ví dụ.

values = {
    'last': 100,
    'first': 1,
    'middle': 50
}
await redis.zadd('sorted_set_1', values) # Sorted set sẽ chứa ["first", "middle", "last"]

Điểm cần lưu ý khi ghi dữ liệu vào Redis

Như đã nhắc đến trong phần trước, Redis lưu dữ liệu dưới dạng bytes. Vì thế, nó sẽ thực hiện các bước chuyển đổi dưới đây trước khi ghi dữ liệu vào thiết bị lưu trữ.

  • Nếu dữ liệu đã có dạng bytes thì dữ liệu đó được ghi nguyên dạng vào Redis.
  • Nếu dữ liệu là string thì string đó được encode sang dạng bytes bằng encoding mặc định. Sau đó giá trị đã encode sẽ được ghi vào Redis.
  • Trong các trường hợp còn lại, dữ liệu được chuyển sang dạng string và được xử lý như ở trường hợp 2 ở trên.

Ví dụ, có thể ta sẽ nghĩ rằng đoạn code dưới đây sẽ ghi số 1 vào Redis.

await redis.set('want_a_number', 1)

Nhưng thực tế Redis sẽ ghi lại giá trị đã encode của string 1.

rs_bytes = await redis.get('want_a_number')
print(typeof(rs_bytes)) # bytes
print(rs_bytes.decode('utf-8')) # "1"

Nếu ta ghi kiểu hash vào Redis thì không chỉ giá trị của trường mà tên của trường cũng sẽ được lưu dưới dạng bytes.

Đọc dữ liệu từ Redis bằng aioredis

Cũng giống như khi ghi dữ liệu, với mỗi kiểu dữ liệu của Redis thì aioredis sẽ có hàm tương ứng để đọc giá trị của chúng. Ta sẽ tái sử dụng các dữ liệu vừa ghi vào Redis trong phần trên làm ví dụ.

Đọc kiểu string từ Redis

Đây vẫn là trường hợp đơn giản nhất, ta cần dùng hàm get.

string_bytes = await redis.get('string_1')
string_value = string_bytes.decode('utf-8')
print(string_value) # "Example"

Hàm này sẽ trả về None nếu key ta dùng không tồn tại. Như đã nhắc đến trong phần tạo kết nối tới Redis, kết nối của ta không tự động decode giá trị trả về từ Redis. Đó là lý do vì sao ta phải tự gọi hàm decode để chuyển bytes sang string.

Đọc kiểu list từ Redis

Để đọc tất cả phần tử giữa 2 index, ta dùng hàm lrange. Chú ý là index ở đây có thể nhận giá trị âm giống như khi ta đặt index cho array trong Python. Đồng thời aioredis có thể tự xử lý IndexError. Giá trị trả về sẽ có dạng array.

members = await redis.lrange('list_4', 1, 3)
print(members)

# Kết quả là:
# [b'three', b'four', b'five']

Còn để đọc giá trị cuối cùng đồng thời xóa giá trị đó khỏi list, ta dùng hàm rpop.

last = await redis.rpop('list_4') # last có giá trị b'six'
                                  # 'list_4' giờ chỉ còn 4 phần tử

Nếu key ta sử dụng không tồn tại thì lrange sẽ trả về một array rỗng, còn rpop sẽ trả về None.

Đọc kiểu hash từ Redis

Để đọc một trường từ hash, ta dùng hàm hget. Nếu key ta dùng hoặc tên trường truyền vào không tồn tại trong hash thì hget sẽ trả về None.

first_name = await redis.hget('hash_1', 'first_name')
print(first_name.decode('utf-8')) # John

Để đọc nhiều trường từ hash, ta dùng hàm hmget. Ta cần truyền tên tất cả các trường muốn đọc vào làm tham số, và kiểu giá trị trả về sẽ là array. Nếu key ta dùng không tồn tại thì giá trị trả về sẽ là một array với tất cả phần tử là None. Còn nếu một trường nào đó không tồn tại thì giá trị tương ứng trong array sẽ là None.

name = await redis.hmget('hash_1', 'first_name', 'last_name')
first_name = name[0].decode('utf-8')
last_name = name[1].decode('utf-8')
print(first_name, last_name) # John Doe

Còn để đọc giá trị tất cả các trường trong hash, ta dùng hàm hgetall. Kiểu giá trị trả về sẽ là dictionary. Nếu key ta dùng không tồn tại thì giá trị trả về sẽ là một dictionary rỗng.

data = await redis.hmget('hash_1')
print(data)

# Kết quả là:
# {b'first_name': b'John', b'last_name': b'Doe', b'age': b'100'}

Như đã nhắc đến ở phần trước, không chỉ giá trị trong dictionary có kiểu bytes mà cả key của dictionary cũng là bytes. Ta cần chú ý điều này khi đọc dữ liệu từ dictionary.

first_name_key = 'first_name'.encode('utf-8')
first_name = data[first_name_key]
print(first_name.decode('utf-8')) # John

Đọc kiểu set từ Redis

Để đọc tất cả các phần tử trong set, ta dùng hàm smembers. Kiểu giá trị trả về là set.

members = await redis.smembers('set_1')

Để đọc nhiều giá trị ngẫu nhiên từ set đồng thời xóa chúng khỏi set, ta gọi hàm spop và truyền vào biến count.

three_members = await redis.spop('set_1', 3) # Đọc ngẫu nhiên 3 phần tử từ set
                                             # 'set_1' sẽ chỉ còn chứa 2 phần tử

Nếu ta chỉ muốn đọc phần tử mà không xóa chúng khỏi set thì ta dùng hàm srandmember.

the_rest = await redis.srandmember('set_1', 2) # Đọc ngẫu nhiên 2 phần tử từ set
                                               # Không phần tử nào bị xóa khỏi 'set_1'

Với hàm smembers, nếu key ta dùng không tồn tại thì giá trị trả về sẽ là set rỗng; còn hàm spop và srandmember sẽ trả về array rỗng. Còn nếu ta muốn đọc nhiều phần tử hơn số phần tử thực tế mà set có thì giá trị trả về vẫn là toàn bộ set đó.

Đọc kiểu sorted set từ Redis

Để đọc tất cả các phần từ nằm giữa 2 vị trí, ta dùng hàm zrange. Cũng giống như với hàm lrange, ta có thể truyền giá trị âm làm index, và aioredis có thể tự động xử lý IndexError. Kiểu giá trị trả về là array.

last_two = await redis.zrange('sorted_set_1', 1, -1)
print(last_two)

# Kết quả là:
# [b'middle', b'last']

Nếu ta muốn sắp xếp các phần tử theo thứ tự giảm dần trước khi đọc dữ liệu thì ta đặt biến desc là True.

last_two = await redis.zrange('sorted_set_1', 1, -1, desc=True)
print(last_two)

# Kết quả là:
# [b'middle', b'first']

Trong cả 2 trường hợp, nếu key ta dùng không tồn tại thì giá trị trả về sẽ là array rỗng.

Dùng pipeline để thực hiện song song nhiều request

Có lẽ các bạn đã nhận ra rằng đoạn code ví dụ dưới đây chưa phải là tối ưu.

await redis.set('string_1', 'Example')
await redis.set('string_2', 'Another Example')

Ta thực hiện 2 request non-blocking mà ta lại đợi request đầu tiên chạy xong rồi mới bắt đầu request thứ 2. Nhưng ta có thể bắt đầu cả 2 request cùng lúc và đợi cho cả 2 kết thúc. Thông thường ta thực hiện điều này bằng asyncio.gather.

tasks = []

first_string = redis.set('string_1', 'Example')
first_string_task = asyncio.ensure_future(first_string)
tasks.append(first_string_task)

second_string = redis.set('string_2', 'Another Example')
second_string_task = asyncio.ensure_future(second_string)
tasks.append(second_string_task)

await asyncio.gather(*tasks)

Tuy nhiên, nếu ta đã biết trước số request muốn gửi thì ta có thể dùng pipeline của aioredis để đơn giản hóa đoạn code ở trên.

async with redis.pipeline() as pipe:
    pipe.set('string_1', 'Example').set('string_2', 'Another Example')
    await pipe.execute()

Thử viết hàm get_or_set

Khi sử dụng cache, tôi thường viết một hàm get_or_set như sau.

  • Nhận vào một key và một delegate.
  • Kiểm tra xem key đó đã tồn tại trong cache hay chưa.
  • Nếu key đó đã tồn tại thì trả về giá trị tương ứng.
  • Còn nếu key chưa tồn tại thì gọi delegate để tạo giá trị mới, ghi giá trị đó vào cache, rồi trả lại giá trị đó.

Với aioredis và Python, ta có thể viết hàm như sau.

async def get_or_set(client, key, delegate):
    # Dùng key để lấy hash từ cache
    val = await client.hgetall(key)

    # Nếu giá trị trả về không phải là dictionary rỗng
    if val:
        # Thì key này đã tồn tại, trả về giá trị của nó
        return val

    # Còn nếu key chưa tồn tại thì gọi delegate để tạo giá trị mới
    val = delegate()
    # Ghi nó vào cache
    await client.hset(key, mapping=val)

    # Và trả về giá trị mới đó
    return val

Hàm ở trên chỉ áp dụng cho dữ liệu với kiểu hash. Ta dùng nó như sau.

import datetime

def load_person():
    return {
        'first_name': 'James',
        'last_name': 'Bonds',
        'age': 100,
        'time_stamp': str(datetime.datetime.now())
    }

rs = await get_or_set(redis, 'hash_get_or_set', load_person)
print(rs)

Dù ta có gọi hàm trên bao nhiêu lần chăng nữa, giá trị của time_stamp cũng không đổi.

{b'first_name': b'James', b'last_name': b'Bonds', b'age': b'100', b'time_stamp': <Thời điểm ta gọi hàm trên lần đầu>}

Kết thúc

Trong bài này, chúng ta đã dùng thử aioredis để đọc và ghi dữ liệu vào Redis server bằng những request non-blocking. Trong phần 2, ta sẽ thực hiện điều tương tự bằng C#.

A software developer from Vietnam and is currently living in Japan.

One Thought on “Kết nối với Redis bằng aioredis và Python”

Leave a Reply