# 存储一个用户对象（flags=0, TTL=3600秒, 数据长度=13字节）
set user:1001 0 3600 13
{"name":"Bob"}

echo "delete user:1001" | nc 127.0.0.1 11211
# DELETED

# 设置计数器
set view_count 0 0 1
0
STORED

# 递增
incr view_count 1
1

incr view_count 10
11

# 递减
decr view_count 5
6

# 正常模式
set key1 0 0 5
hello
STORED

# noreply 模式（无响应）
set key2 0 0 5 noreply
hello

import socket

def batch_set_noreply(sock, items: dict, ttl: int = 300):
    """批量写入，使用 noreply 提升吞吐量"""
    for key, value in items.items():
        data = value.encode() if isinstance(value, str) else value
        cmd = f"set {key} 0 {ttl} {len(data)} noreply\r\n"
        sock.sendall(cmd.encode() + data + b"\r\n")
    # 不需要等待响应，直接返回

# 使用
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('127.0.0.1', 11211))
batch_set_noreply(sock, {
    "cache:a": "value_a",
    "cache:b": "value_b",
    "cache:c": "value_c",
})

#!/usr/bin/env python3
"""memcached_text_parser.py — Memcached 文本协议解析器"""

from dataclasses import dataclass
from typing import List, Optional, Union
from enum import Enum
import socket

class ResponseType(Enum):
    STORED = "STORED"
    NOT_STORED = "NOT_STORED"
    EXISTS = "EXISTS"
    NOT_FOUND = "NOT_FOUND"
    DELETED = "DELETED"
    ERROR = "ERROR"
    VALUE = "VALUE"
    END = "END"

@dataclass
class ValueResponse:
    key: str
    flags: int
    cas_unique: Optional[int]
    data: bytes

@dataclass
class Response:
    type: ResponseType
    values: List[ValueResponse]
    message: str = ""

class MemcachedTextParser:
    """Memcached 文本协议解析器"""

    def __init__(self):
        self.buffer = b""

    def feed(self, data: bytes):
        self.buffer += data

    def parse_response(self, command_type: str = "get") -> Optional[Response]:
        """尝试从 buffer 中解析完整响应"""
        lines = self.buffer.split(b"\r\n")

        if command_type in ("get", "gets"):
            return self._parse_value_response(lines)
        elif command_type in ("set", "add", "replace", "append", "prepend", "cas"):
            return self._parse_store_response(lines)
        elif command_type == "delete":
            return self._parse_simple_response(lines)
        elif command_type in ("incr", "decr"):
            return self._parse_counter_response(lines)
        return None

    def _parse_value_response(self, lines: list) -> Optional[Response]:
        """解析 VALUE 响应"""
        if b"END" not in lines:
            return None  # 数据不完整

        values = []
        i = 0
        while i < len(lines):
            line = lines[i]
            if line == b"END":
                return Response(type=ResponseType.END, values=values)
            if line.startswith(b"VALUE "):
                parts = line.decode().split()
                key = parts[1]
                flags = int(parts[2])
                length = int(parts[3])
                cas = int(parts[5]) if len(parts) > 5 and parts[4] == "cas" else None

                i += 1
                if i < len(lines):
                    data = lines[i]
                    values.append(ValueResponse(
                        key=key, flags=flags,
                        cas_unique=cas, data=data
                    ))
            i += 1

        return None  # 数据不完整

    def _parse_store_response(self, lines: list) -> Optional[Response]:
        """解析存储命令响应"""
        if len(lines) < 2:
            return None
        first_line = lines[0]
        if first_line == b"STORED":
            return Response(type=ResponseType.STORED, values=[])
        elif first_line == b"NOT_STORED":
            return Response(type=ResponseType.NOT_STORED, values=[])
        elif first_line == b"EXISTS":
            return Response(type=ResponseType.EXISTS, values=[])
        elif first_line == b"NOT_FOUND":
            return Response(type=ResponseType.NOT_FOUND, values=[])
        elif first_line.startswith(b"CLIENT_ERROR"):
            return Response(type=ResponseType.ERROR,
                          values=[], message=first_line.decode())
        elif first_line.startswith(b"SERVER_ERROR"):
            return Response(type=ResponseType.ERROR,
                          values=[], message=first_line.decode())
        elif first_line == b"ERROR":
            return Response(type=ResponseType.ERROR, values=[])
        return None

    def _parse_simple_response(self, lines: list) -> Optional[Response]:
        """解析简单响应"""
        if len(lines) < 2:
            return None
        line = lines[0]
        if line == b"DELETED":
            return Response(type=ResponseType.DELETED, values=[])
        elif line == b"NOT_FOUND":
            return Response(type=ResponseType.NOT_FOUND, values=[])
        return None

    def _parse_counter_response(self, lines: list) -> Optional[Response]:
        """解析计数器响应"""
        if len(lines) < 2:
            return None
        line = lines[0]
        if line == b"NOT_FOUND":
            return Response(type=ResponseType.NOT_FOUND, values=[])
        try:
            value = int(line)
            return Response(type=ResponseType.STORED,
                          values=[], message=str(value))
        except ValueError:
            return None


# 测试
parser = MemcachedTextParser()
parser.feed(b"VALUE user:1001 0 13\r\n")
parser.feed(b'{"name":"Bob"}\r\n')
parser.feed(b"END\r\n")

result = parser.parse_response("get")
if result and result.values:
    print(f"Key: {result.values[0].key}")
    print(f"Data: {result.values[0].data.decode()}")

import json
import uuid
import socket

class SessionStore:
    def __init__(self, host='127.0.0.1', port=11211):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((host, port))

    def create_session(self, user_data: dict, ttl: int = 1800) -> str:
        session_id = str(uuid.uuid4())
        data = json.dumps(user_data).encode()
        self.sock.sendall(
            f"set session:{session_id} 0 {ttl} {len(data)}\r\n".encode()
            + data + b"\r\n"
        )
        self.sock.recv(1024)
        return session_id

    def get_session(self, session_id: str) -> dict:
        self.sock.sendall(f"get session:{session_id}\r\n".encode())
        resp = self.sock.recv(65536).decode()
        lines = resp.strip().split("\r\n")
        if lines[0].startswith("VALUE"):
            return json.loads(lines[1])
        return None

    def destroy_session(self, session_id: str) -> bool:
        self.sock.sendall(f"delete session:{session_id}\r\n".encode())
        return b"DELETED" in self.sock.recv(1024)

# 使用
store = SessionStore()
sid = store.create_session({"user_id": 1001, "role": "admin"})
print(f"Session ID: {sid}")
print(f"Session Data: {store.get_session(sid)}")

def decrement_stock(sock, product_id: str, quantity: int) -> bool:
    """
    使用 DECR 原子扣减库存。
    注意：此方案要求库存值在缓存中初始化。
    """
    key = f"stock:{product_id}"
    sock.sendall(f"decr {key} {quantity}\r\n".encode())
    resp = sock.recv(1024).decode().strip()

    if resp == "NOT_FOUND":
        return False  # 库存键不存在

    remaining = int(resp)
    if remaining < 0:
        # 库存不足，回滚
        sock.sendall(f"incr {key} {quantity}\r\n".encode())
        sock.recv(1024)
        return False

    return True  # 扣减成功