使用 Redis 流實現訊息佇列的程式碼
阿新 • • 發佈:2020-01-09
在介紹了 Redis 流的基本功能之後, 現在是時候使用這些功能來構建一些實際的應用了。 訊息佇列作為流的典型應用之一, 具有非常好的示範性, 因此我們將使用 Redis 流的相關功能構建一個訊息佇列應用, 這個訊息佇列跟我們之前使用其他 Redis 資料結構構建的訊息佇列具有相似的功能。
程式碼清單 10-1 展示了一個具有基本功能的訊息佇列實現:
- 程式碼最開頭的是幾個轉換函式, 它們負責對程式的相關輸入輸出進行轉換和格式化;
- MessageQueue 類用於實現訊息佇列, 它的新增訊息、移除訊息以及返回訊息數量三個方法分別使用了流的XADD 命令、XDEL 命令和XLEN 命令;
- 訊息佇列的兩個獲取方法get_message() 和get_by_range() 分別以兩種形式呼叫了流的XRANGE 命令;
- 最後, 用於迭代訊息的iterate() 方法使用了XREAD 命令對流進行迭代。
程式碼清單 10-1 使用 Redis 流實現的訊息佇列:/stream/message_queue.py
def reconstruct_message_list(message_list): """ 為了讓多條訊息能夠以更結構化的方式返回給呼叫者, 將 Redis 返回的多條訊息從原來的格式: [(id1,{k1:v1,k2:v2,...}),(id2,...] 轉換成以下格式: [{id1: {k1:v1,...}},{id2: {k1:v1,...] """ result = [] for id,kvs in message_list: result.append({id: kvs}) return result def get_message_from_nested_list(lst): """ 從巢狀列表中取出訊息本體。 """ return lst[0][1] class MessageQueue: """ 使用 Redis 流實現的訊息佇列。 """ def __init__(self,client,stream_key): self.client = client self.stream = stream_key def add_message(self,key_value_pairs): """ 將給定的鍵值對存入到訊息裡面,並返回相應的訊息 ID 。 """ return self.client.xadd(self.stream,key_value_pairs) def get_message(self,message_id): """ 根據給定的訊息 ID 返回相應的訊息,如果訊息不存在則返回 None 。 """ reply = self.client.xrange(self.stream,message_id,message_id) if len(reply) == 1: return get_message_from_nested_list(reply) def remove_message(self,message_id): """ 根據給定的訊息 ID 刪除相應的訊息,如果訊息不存在則忽略該動作。 """ self.client.xdel(self.stream,message_id) def len(self): """ 返回訊息佇列的長度。 """ return self.client.xlen(self.stream) def get_by_range(self,start_id,end_id,max_item=10): """ 根據給定的 ID 區間範圍返回佇列中的訊息。 """ reply = self.client.xrange(self.stream,max_item) return reconstruct_message_list(reply) def iterate(self,start_id=0,max_item=10): """ 對訊息佇列進行迭代,返回最多 N 條大於給定 ID 的訊息。 """ reply = self.client.xread({self.stream: start_id},max_item) if len(reply) == 0: return list() else: messages = get_message_from_nested_list(reply) return reconstruct_message_list(messages)
對於這個訊息佇列實現, 我們可以通過執行以下程式碼, 創建出它的例項:
>>> from redis import Redis >>> from message_queue import MessageQueue >>> client = Redis(decode_responses=True) >>> mq = MessageQueue(client,"mq")
然後通過執行以下程式碼, 向佇列裡面新增十條訊息:
>>> for i in range(10): ... key = "key{0}".format(i) ... value = "value{0}".format(i) ... msg = {key:value} ... mq.add_message(msg) ... '1554113926280-0' '1554113926280-1' '1554113926281-0' '1554113926281-1' '1554113926281-2' '1554113926281-3' '1554113926281-4' '1554113926281-5' '1554113926281-6' '1554113926282-0'
還可以根據 ID 獲取指定的訊息, 又或者使用get_by_range() 方法同時獲取多條訊息:
>>> mq.get_message('1554113926280-0') {'key0': 'value0'} >>> mq.get_message('1554113926280-1') {'key1': 'value1'} >>> mq.get_by_range("-","+",3) [{'1554113926280-0': {'key0': 'value0'}},{'1554113926280-1': {'key1': 'value1'}},{'1554113926281-0': {'key2': 'value2'}}]
又或者使用iterate() 方法對訊息佇列進行迭代, 等等:
>>> mq.iterate(0,{'1554113926281-0': {'key2': 'value2'}}] >>> mq.iterate('1554113926281-0',3) [{'1554113926281-1': {'key3': 'value3'}},{'1554113926281-2': {'key4': 'value4'}},{'1554113926281-3': {'key5': 'value5'}}]
總結
以上所述是小編給大家介紹的使用 Redis 流實現訊息佇列的程式碼,希望對大家有所幫助,如果大家有任何疑問請給我留言,小編會及時回覆大家的。在此也非常感謝大家對我們網站的支援!
如果你覺得本文對你有幫助,歡迎轉載,煩請註明出處,謝謝!