# 文本协议解析伪代码
line = socket.readline()          # 读取一行
parts = line.split(" ")           # 按空格分割
command = parts[0]                # 提取命令
key = parts[1]                    # 提取 key
flags = int(parts[2])             # 转换 flags
exptime = int(parts[3])           # 转换过期时间
length = int(parts[4])            # 提取数据长度
data = socket.read(length)        # 读取数据体

# 二进制协议解析伪代码
header = socket.read(24)          # 固定 24 字节头
magic = header[0]                 # 直接字节读取
opcode = header[1]
key_length = struct.unpack(">H", header[2:4])[0]
body_length = struct.unpack(">I", header[8:12])[0]
body = socket.read(body_length)   # 读取完整 body

memcached \
  -p 11211 \       # 监听端口
  -m 1024 \        # 最大内存（MB）
  -c 10240 \       # 最大连接数
  -t 4 \           # 工作线程数
  -l 127.0.0.1 \   # 监听地址
  -U 0 \           # 禁用 UDP（0=禁用）
  -o modern \      # 启用现代特性（1.6+）
  -v               # 详细日志（-vv 更详细）

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('127.0.0.1', 11211))

# 管道化发送多个命令
sock.sendall(b"set key1 0 0 5\r\nhello\r\n")
sock.sendall(b"set key2 0 0 5\r\nworld\r\n")
sock.sendall(b"get key1\r\n")

# 按顺序接收响应
print(sock.recv(4096).decode())
# STORED\r\nSTORED\r\nVALUE key1 0 5\r\nhello\r\nEND\r\n

# 连接到 Memcached
telnet 127.0.0.1 11211

# 设置
echo -e "set testkey 0 300 4\r\ntest\r\n" | nc 127.0.0.1 11211

# 获取
echo "get testkey" | nc 127.0.0.1 11211

# 统计
echo "stats" | nc 127.0.0.1 11211

#!/usr/bin/env python3
"""memcached_demo.py — Memcached 文本协议快速演示"""

import socket

def send_command(sock, command: str, data: bytes = b"") -> str:
    """发送命令并接收响应"""
    sock.sendall(command.encode() + b"\r\n")
    if data:
        sock.sendall(data + b"\r\n")

    response = b""
    while True:
        chunk = sock.recv(4096)
        response += chunk
        if b"END\r\n" in response or b"STORED\r\n" in response \
           or b"NOT_STORED\r\n" in response or b"DELETED\r\n" in response \
           or b"NOT_FOUND\r\n" in response:
            break
    return response.decode().strip()

def main():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(('127.0.0.1', 11211))

    # 1. 查看版本
    print("=== Version ===")
    sock.sendall(b"version\r\n")
    print(sock.recv(1024).decode().strip())

    # 2. 存储数据
    print("\n=== Set ===")
    result = send_command(sock, "set user:1001 0 3600 13", b'{"name":"Bob"}')
    print(result)

    # 3. 读取数据
    print("\n=== Get ===")
    result = send_command(sock, "get user:1001")
    print(result)

    # 4. 删除数据
    print("\n=== Delete ===")
    result = send_command(sock, "delete user:1001")
    print(result)

    # 5. 查看统计
    print("\n=== Stats (摘要) ===")
    sock.sendall(b"stats\r\n")
    data = sock.recv(8192).decode()
    for line in data.split("\r\n"):
        if line.startswith("STAT") and any(
            k in line for k in ["curr_items", "total_items", "bytes", "cmd_get", "cmd_set"]
        ):
            print(line)

    sock.sendall(b"quit\r\n")
    sock.close()

if __name__ == "__main__":
    main()

import json
import socket
from typing import Optional, Dict, Any

class SimpleMemcachedClient:
    def __init__(self, host: str = '127.0.0.1', port: int = 11211):
        self.host = host
        self.port = port
        self.sock = None
        self._connect()

    def _connect(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((self.host, self.port))

    def set(self, key: str, value: Any, ttl: int = 300) -> bool:
        data = json.dumps(value).encode()
        cmd = f"set {key} 0 {ttl} {len(data)}\r\n"
        self.sock.sendall(cmd.encode() + data + b"\r\n")
        resp = self.sock.recv(1024).decode().strip()
        return resp == "STORED"

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

    def close(self):
        self.sock.sendall(b"quit\r\n")
        self.sock.close()

# 使用示例
cache = SimpleMemcachedClient()

# 缓存商品信息
product = {
    "id": 10001,
    "name": "机械键盘",
    "price": 599.0,
    "stock": 100
}
cache.set("product:10001", product, ttl=600)

# 查询商品
cached = cache.get("product:10001")
if cached:
    print(f"命中缓存: {cached['name']} ¥{cached['price']}")
else:
    print("缓存未命中，查询数据库...")