使用Python編寫自己的微型Redis

更新時間：2024年03月12日 14:15:59 作者：Python魔法師

這篇文章主要為大家詳細介紹了使用Python編寫自己的微型Redis的相關知識,文中的示例代碼講解詳細,具有一定的學習價值,感興趣的小伙伴可以了解一下

前幾天我想到,寫一個簡單的東西會很整潔雷迪斯-像數據庫服務器。雖然我有很多 WSGI應用程序的經驗,數據庫服務器展示了一種新穎挑戰(zhàn),并被證明是學習如何工作的不錯的實際方法 Python中的套接字。在這篇文章中,我將分享我在此過程中學到的知識。

我項目的目的是編寫一個簡單的服務器我可以用我的任務隊列項目稱為 Huey。 Huey使用Redis作為默認存儲引擎來跟蹤被引用的工作, 完成的工作和其他結果。就本職位而言, 我進一步縮小了原始項目的范圍,以免造成混亂使用代碼的水域,您可以很容易地自己寫,但是如果您很好奇,你可以看看最終結果這里 (文件)。

我們將要構建的服務器將能夠響應以下命令:

GET <key>
SET <key> <value>
DELETE <key>
FLUSH
MGET <key1> ... <keyn>
MSET <key1> <value1> ... <keyn> <valuen>

我們還將支持以下數據類型:

Strings and Binary Data
Numbers
NULL
Arrays (which may be nested)
Dictionaries (which may be nested)
Error messages

為了異步處理多個客戶端,我們將使用 gevent, 但是您也可以使用標準庫的 SocketServer 模塊與要么 ForkingMixin 或 ThreadingMixin。

骨架

讓我們?yōu)榉掌髟O置一個框架。我們需要服務器本身,以及新客戶端連接時要執(zhí)行的回調。另外,我們需要某種邏輯來處理客戶端請求并發(fā)送響應。

這是一個開始:

from gevent import socket
from gevent.pool import Pool
from gevent.server import StreamServer

from collections import namedtuple
from io import BytesIO
from socket import error as socket_error


# We'll use exceptions to notify the connection-handling loop of problems.
class CommandError(Exception): pass
class Disconnect(Exception): pass

Error = namedtuple('Error', ('message',))


class ProtocolHandler(object):
    def handle_request(self, socket_file):
        # Parse a request from the client into it's component parts.
        pass

    def write_response(self, socket_file, data):
        # Serialize the response data and send it to the client.
        pass


class Server(object):
    def __init__(self, host='127.0.0.1', port=31337, max_clients=64):
        self._pool = Pool(max_clients)
        self._server = StreamServer(
            (host, port),
            self.connection_handler,
            spawn=self._pool)

        self._protocol = ProtocolHandler()
        self._kv = {}

    def connection_handler(self, conn, address):
        # Convert "conn" (a socket object) into a file-like object.
        socket_file = conn.makefile('rwb')

        # Process client requests until client disconnects.
        while True:
            try:
                data = self._protocol.handle_request(socket_file)
            except Disconnect:
                break

            try:
                resp = self.get_response(data)
            except CommandError as exc:
                resp = Error(exc.args[0])

            self._protocol.write_response(socket_file, resp)

    def get_response(self, data):
        # Here we'll actually unpack the data sent by the client, execute the
        # command they specified, and pass back the return value.
        pass

    def run(self):
        self._server.serve_forever()

希望以上代碼相當清楚。我們已經分開了擔憂,以便協議處理屬于自己的類,有兩種公共方法: handle_request 和 write_response。服務器本身使用協議處理程序解壓縮客戶端請求并將服務器響應序列化回客戶。The get_response() 該方法將用于執(zhí)行命令由客戶發(fā)起。

仔細查看代碼 connection_handler() 方法,你可以看到我們在套接字對象周圍獲得了類似文件的包裝紙。這個包裝器讓我們抽象一些怪癖通常會遇到使用原始插座的情況。函數輸入無窮循環(huán),讀取客戶端的請求,發(fā)送響應,最后客戶端斷開連接時退出循環(huán)(由 read() 返回一個空字符串)。

我們使用鍵入的異常來處理客戶端斷開連接并通知用戶錯誤處理命令。例如,如果用戶做錯了對服務器的格式化請求,我們將提出一個 CommandError, 哪個是序列化為錯誤響應并發(fā)送給客戶端。

在繼續(xù)之前,讓我們討論客戶端和服務器將如何通信。

線程

我面臨的第一個挑戰(zhàn)是如何處理通過線。我在網上找到的大多數示例都是毫無意義的回聲服務器,它們進行了轉換套接字到類似文件的對象,并且剛剛調用 readline()。如果我想用新線存儲一些腌制的數據或字符串,我需要一些一種序列化格式。

在浪費時間嘗試發(fā)明合適的東西之后,我決定閱讀有關文檔 Redis協議, 其中事實證明實施起來非常簡單,并且具有支持幾種不同的數據類型。

Redis協議使用請求/響應通信模式與客戶。來自服務器的響應將使用第一個字節(jié)來指示數據類型,然后是數據,以回車/線路進給終止。

讓我們填寫協議處理程序的類,使其實現Redis 協議。

class ProtocolHandler(object):
    def __init__(self):
        self.handlers = {
            '+': self.handle_simple_string,
            '-': self.handle_error,
            ':': self.handle_integer,
            '$': self.handle_string,
            '*': self.handle_array,
            '%': self.handle_dict}

    def handle_request(self, socket_file):
        first_byte = socket_file.read(1)
        if not first_byte:
            raise Disconnect()

        try:
            # Delegate to the appropriate handler based on the first byte.
            return self.handlers[first_byte](socket_file)
        except KeyError:
            raise CommandError('bad request')

    def handle_simple_string(self, socket_file):
        return socket_file.readline().rstrip('\r\n')

    def handle_error(self, socket_file):
        return Error(socket_file.readline().rstrip('\r\n'))

    def handle_integer(self, socket_file):
        return int(socket_file.readline().rstrip('\r\n'))

    def handle_string(self, socket_file):
        # First read the length ($<length>\r\n).
        length = int(socket_file.readline().rstrip('\r\n'))
        if length == -1:
            return None  # Special-case for NULLs.
        length += 2  # Include the trailing \r\n in count.
        return socket_file.read(length)[:-2]

    def handle_array(self, socket_file):
        num_elements = int(socket_file.readline().rstrip('\r\n'))
        return [self.handle_request(socket_file) for _ in range(num_elements)]

    def handle_dict(self, socket_file):
        num_items = int(socket_file.readline().rstrip('\r\n'))
        elements = [self.handle_request(socket_file)
                    for _ in range(num_items * 2)]
        return dict(zip(elements[::2], elements[1::2]))

對于協議的序列化方面,我們將執(zhí)行與上述相反的操作: 將Python對象轉換為序列化的對象!

class ProtocolHandler(object):
    # ... above methods omitted ...
    def write_response(self, socket_file, data):
        buf = BytesIO()
        self._write(buf, data)
        buf.seek(0)
        socket_file.write(buf.getvalue())
        socket_file.flush()

    def _write(self, buf, data):
        if isinstance(data, str):
            data = data.encode('utf-8')

        if isinstance(data, bytes):
            buf.write('$%s\r\n%s\r\n' % (len(data), data))
        elif isinstance(data, int):
            buf.write(':%s\r\n' % data)
        elif isinstance(data, Error):
            buf.write('-%s\r\n' % error.message)
        elif isinstance(data, (list, tuple)):
            buf.write('*%s\r\n' % len(data))
            for item in data:
                self._write(buf, item)
        elif isinstance(data, dict):
            buf.write('%%%s\r\n' % len(data))
            for key in data:
                self._write(buf, key)
                self._write(buf, data[key])
        elif data is None:
            buf.write('$-1\r\n')
        else:
            raise CommandError('unrecognized type: %s' % type(data))

將協議處理保持在其自己的類中的另一個好處是我們可以重復使用 handle_request 和 write_response 建立方法客戶端庫。

執(zhí)行命令

類Server 我們模擬的課程現在需要 get_response() 方法已實施。命令將假定由客戶端以簡單方式發(fā)送字符串或命令參數數組,因此 data 傳遞給 get_response() 將是字節(jié)或列表。為了簡化處理,如果 data 這是一個簡單的字符串,我們將通過拆分將其轉換為列表空格。

第一個參數將是命令名稱,并帶有任何其他參數屬于指定命令。就像我們對第一個的映射一樣字節(jié)給處理者 ProtocolHandler, 讓我們創(chuàng)建一個的映射命令回調 Server:

class Server(object):
    def __init__(self, host='127.0.0.1', port=31337, max_clients=64):
        self._pool = Pool(max_clients)
        self._server = StreamServer(
            (host, port),
            self.connection_handler,
            spawn=self._pool)

        self._protocol = ProtocolHandler()
        self._kv = {}

        self._commands = self.get_commands()

    def get_commands(self):
        return {
            'GET': self.get,
            'SET': self.set,
            'DELETE': self.delete,
            'FLUSH': self.flush,
            'MGET': self.mget,
            'MSET': self.mset}

    def get_response(self, data):
        if not isinstance(data, list):
            try:
                data = data.split()
            except:
                raise CommandError('Request must be list or simple string.')

        if not data:
            raise CommandError('Missing command')

        command = data[0].upper()
        if command not in self._commands:
            raise CommandError('Unrecognized command: %s' % command)

        return self._commands[command](*data[1:])

我們的服務器快完成了! 我們只需要執(zhí)行六個命令在 get_commands() 方法:

class Server(object):
    def get(self, key):
        return self._kv.get(key)

    def set(self, key, value):
        self._kv[key] = value
        return 1

    def delete(self, key):
        if key in self._kv:
            del self._kv[key]
            return 1
        return 0

    def flush(self):
        kvlen = len(self._kv)
        self._kv.clear()
        return kvlen

    def mget(self, *keys):
        return [self._kv.get(key) for key in keys]

    def mset(self, *items):
        data = zip(items[::2], items[1::2])
        for key, value in data:
            self._kv[key] = value
        return len(data)

而已! 我們的服務器現在可以開始處理請求了。在下一個本節(jié),我們將實現一個客戶端與服務器進行交互。

客戶端

要與服務器交互,讓我們重新使用 ProtocolHandler 類到實現一個簡單的客戶端?？蛻舳藢⑦B接到服務器并發(fā)送命令編碼為列表。我們將同時使用 write_response() 和 handle_request() 編碼請求和處理服務器響應的邏輯分別。

class Client(object):
    def __init__(self, host='127.0.0.1', port=31337):
        self._protocol = ProtocolHandler()
        self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._socket.connect((host, port))
        self._fh = self._socket.makefile('rwb')

    def execute(self, *args):
        self._protocol.write_response(self._fh, args)
        resp = self._protocol.handle_request(self._fh)
        if isinstance(resp, Error):
            raise CommandError(resp.message)
        return resp

與 execute() 方法上,我們可以傳遞任意參數列表,這些參數將被編碼為數組并發(fā)送到服務器。來自服務器的響應被解析并作為Python對象返回。為了方便起見,我們可以為各個命令編寫客戶端方法:

class Client(object):
    # ...
    def get(self, key):
        return self.execute('GET', key)

    def set(self, key, value):
        return self.execute('SET', key, value)

    def delete(self, key):
        return self.execute('DELETE', key)

    def flush(self):
        return self.execute('FLUSH')

    def mget(self, *keys):
        return self.execute('MGET', *keys)

    def mset(self, *items):
        return self.execute('MSET', *items)

為了測試我們的客戶端,讓我們配置Python腳本以啟動服務器直接從命令行執(zhí)行時:

測試服務器

要測試服務器,只需從命令行執(zhí)行服務器的Python模塊即可。在另一個終端中,打開Python解釋器并導入 Client 來自服務器模塊的類。安裝客戶端將打開連接,您可以開始運行命令!

>>> from server_ex import Client
>>> client = Client()
>>> client.mset('k1', 'v1', 'k2', ['v2-0', 1, 'v2-2'], 'k3', 'v3')
3
>>> client.get('k2')
['v2-0', 1, 'v2-2']
>>> client.mget('k3', 'k1')
['v3', 'v1']
>>> client.delete('k1')
1
>>> client.get('k1')
>>> client.delete('k1')
0
>>> client.set('kx', {'vx': {'vy': 0, 'vz': [1, 2, 3]}})
1
>>> client.get('kx')
{'vx': {'vy': 0, 'vz': [1, 2, 3]}}
>>> client.flush()
2

完整代碼

from gevent import socket
from gevent.pool import Pool
from gevent.server import StreamServer

from collections import namedtuple
from io import BytesIO
from socket import error as socket_error
import logging


logger = logging.getLogger(__name__)


class CommandError(Exception): pass
class Disconnect(Exception): pass

Error = namedtuple('Error', ('message',))


class ProtocolHandler(object):
    def __init__(self):
        self.handlers = {
            '+': self.handle_simple_string,
            '-': self.handle_error,
            ':': self.handle_integer,
            '$': self.handle_string,
            '*': self.handle_array,
            '%': self.handle_dict}

    def handle_request(self, socket_file):
        first_byte = socket_file.read(1)
        if not first_byte:
            raise Disconnect()

        try:
            # Delegate to the appropriate handler based on the first byte.
            return self.handlers[first_byte](socket_file)
        except KeyError:
            raise CommandError('bad request')

    def handle_simple_string(self, socket_file):
        return socket_file.readline().rstrip('\r\n')

    def handle_error(self, socket_file):
        return Error(socket_file.readline().rstrip('\r\n'))

    def handle_integer(self, socket_file):
        return int(socket_file.readline().rstrip('\r\n'))

    def handle_string(self, socket_file):
        # First read the length ($<length>\r\n).
        length = int(socket_file.readline().rstrip('\r\n'))
        if length == -1:
            return None  # Special-case for NULLs.
        length += 2  # Include the trailing \r\n in count.
        return socket_file.read(length)[:-2]

    def handle_array(self, socket_file):
        num_elements = int(socket_file.readline().rstrip('\r\n'))
        return [self.handle_request(socket_file) for _ in range(num_elements)]
    
    def handle_dict(self, socket_file):
        num_items = int(socket_file.readline().rstrip('\r\n'))
        elements = [self.handle_request(socket_file)
                    for _ in range(num_items * 2)]
        return dict(zip(elements[::2], elements[1::2]))

    def write_response(self, socket_file, data):
        buf = BytesIO()
        self._write(buf, data)
        buf.seek(0)
        socket_file.write(buf.getvalue())
        socket_file.flush()

    def _write(self, buf, data):
        if isinstance(data, str):
            data = data.encode('utf-8')

        if isinstance(data, bytes):
            buf.write('$%s\r\n%s\r\n' % (len(data), data))
        elif isinstance(data, int):
            buf.write(':%s\r\n' % data)
        elif isinstance(data, Error):
            buf.write('-%s\r\n' % error.message)
        elif isinstance(data, (list, tuple)):
            buf.write('*%s\r\n' % len(data))
            for item in data:
                self._write(buf, item)
        elif isinstance(data, dict):
            buf.write('%%%s\r\n' % len(data))
            for key in data:
                self._write(buf, key)
                self._write(buf, data[key])
        elif data is None:
            buf.write('$-1\r\n')
        else:
            raise CommandError('unrecognized type: %s' % type(data))


class Server(object):
    def __init__(self, host='127.0.0.1', port=31337, max_clients=64):
        self._pool = Pool(max_clients)
        self._server = StreamServer(
            (host, port),
            self.connection_handler,
            spawn=self._pool)

        self._protocol = ProtocolHandler()
        self._kv = {}

        self._commands = self.get_commands()

    def get_commands(self):
        return {
            'GET': self.get,
            'SET': self.set,
            'DELETE': self.delete,
            'FLUSH': self.flush,
            'MGET': self.mget,
            'MSET': self.mset}

    def connection_handler(self, conn, address):
        logger.info('Connection received: %s:%s' % address)
        # Convert "conn" (a socket object) into a file-like object.
        socket_file = conn.makefile('rwb')

        # Process client requests until client disconnects.
        while True:
            try:
                data = self._protocol.handle_request(socket_file)
            except Disconnect:
                logger.info('Client went away: %s:%s' % address)
                break

            try:
                resp = self.get_response(data)
            except CommandError as exc:
                logger.exception('Command error')
                resp = Error(exc.args[0])

            self._protocol.write_response(socket_file, resp)

    def run(self):
        self._server.serve_forever()

    def get_response(self, data):
        if not isinstance(data, list):
            try:
                data = data.split()
            except:
                raise CommandError('Request must be list or simple string.')

        if not data:
            raise CommandError('Missing command')

        command = data[0].upper()
        if command not in self._commands:
            raise CommandError('Unrecognized command: %s' % command)
        else:
            logger.debug('Received %s', command)

        return self._commands[command](*data[1:])

    def get(self, key):
        return self._kv.get(key)

    def set(self, key, value):
        self._kv[key] = value
        return 1

    def delete(self, key):
        if key in self._kv:
            del self._kv[key]
            return 1
        return 0

    def flush(self):
        kvlen = len(self._kv)
        self._kv.clear()
        return kvlen

    def mget(self, *keys):
        return [self._kv.get(key) for key in keys]

    def mset(self, *items):
        data = zip(items[::2], items[1::2])
        for key, value in data:
            self._kv[key] = value
        return len(data)


class Client(object):
    def __init__(self, host='127.0.0.1', port=31337):
        self._protocol = ProtocolHandler()
        self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._socket.connect((host, port))
        self._fh = self._socket.makefile('rwb')

    def execute(self, *args):
        self._protocol.write_response(self._fh, args)
        resp = self._protocol.handle_request(self._fh)
        if isinstance(resp, Error):
            raise CommandError(resp.message)
        return resp

    def get(self, key):
        return self.execute('GET', key)

    def set(self, key, value):
        return self.execute('SET', key, value)

    def delete(self, key):
        return self.execute('DELETE', key)

    def flush(self):
        return self.execute('FLUSH')

    def mget(self, *keys):
        return self.execute('MGET', *keys)

    def mset(self, *items):
        return self.execute('MSET', *items)


if __name__ == '__main__':
    from gevent import monkey; monkey.patch_all()
    logger.addHandler(logging.StreamHandler())
    logger.setLevel(logging.DEBUG)
    Server().run()

以上就是使用Python編寫自己的微型Redis的詳細內容，更多關于Python編寫微型Redis的資料請關注腳本之家其它相關文章！

您可能感興趣的文章:

Python
Redis

Pandas reindex重置索引的使用
本文主要介紹了Pandas reindex重置索引的使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2022-06-06
將pytorch轉成longtensor的簡單方法
今天小編就為大家分享一篇將pytorch轉成longtensor的簡單方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2020-02-02
Python Pandas處理CSV文件的常用技巧分享
這篇文章主要和大家分享幾個Python Pandas中處理CSV文件的常用技巧，如：統計列值出現的次數、篩選特定列值、遍歷數據行等，需要的可以參考一下
2022-06-06
Python使用markconv實現將Markdown轉換為HTML
將Markdown轉換為HTML是markconv最核心的功能之一,也是markconv開發(fā)的第一個功能,現在讓我們深入探討如何使用markconv將Markdown文檔轉換為HTML格式吧
2026-03-03
Python類型提示與靜態(tài)類型檢查的高級應用實例代碼
隨著Python在企業(yè)級應用中的地位日益提升,靜態(tài)類型檢查器如 Pyright 正在扮演著越來越重要的角色,這篇文章主要介紹了Python類型提示與靜態(tài)類型檢查的高級應用,文中通過代碼介紹的非常詳細,需要的朋友可以參考下
2026-04-04
Python中逗號轉為空格的三種方法
本文介紹了Python中將逗號轉換為空格的三種方法,包含使用replace函數、使用split函數、使用正則表達式,具有一定的參考價值,感興趣的可以了解一下
2024-02-02
Python使用UDP實現720p視頻傳輸的操作
這篇文章主要介紹了Python使用UDP實現720p視頻傳輸的操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2021-04-04
Python使用openpyxl與pandas讀取Excel文件的對比詳解
在 Python 數據處理與分析場景中,Excel 文件作為經典的數據存儲格式,其讀取與操作效率直接影響開發(fā)流程,openpyxl具備精細化操作單元格、樣式設置、圖表編輯等底層能力；而 pandas 依托強大的數據結構與矢量化運算,本文對兩種工具進行全方位對比
2025-12-12
Selenium chrome配置代理Python版的方法
這篇文章主要介紹了Selenium chrome配置代理Python版的方法，小編覺得挺不錯的，現在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
2018-11-11
使用Python實現不同需求的排行榜功能
這篇文章主要為大家介紹了Python實現不同需求的排行榜功能,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪
2024-01-01