import socket
import selectors

class SimpleRedisProxy:
    def __init__(self, backends):
        """
        backends: [(host, port), ...] Redis 后端列表
        """
        self.backends = backends
        self.sel = selectors.DefaultSelector()

    def start(self, listen_host, listen_port):
        """启动代理服务"""
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((listen_host, listen_port))
        server.listen(100)
        server.setblocking(False)
        self.sel.register(server, selectors.EVENT_READ, self._accept)

        while True:
            events = self.sel.select()
            for key, mask in events:
                callback = key.data
                callback(key.fileobj)

    def _accept(self, sock):
        """接受新连接"""
        client, addr = sock.accept()
        client.setblocking(False)
        backend = self._select_backend(client)
        self.sel.register(client, selectors.EVENT_READ,
                         lambda s: self._relay(s, backend))

    def _select_backend(self, client_sock):
        """选择后端（简单轮询）"""
        import random
        return self._connect_backend(random.choice(self.backends))

    def _connect_backend(self, addr):
        """连接后端"""
        sock = socket.create_connection(addr)
        sock.setblocking(False)
        return sock

    def _relay(self, client_sock, backend_sock):
        """转发数据"""
        try:
            data = client_sock.recv(65536)
            if data:
                backend_sock.sendall(data)
            else:
                self.sel.unregister(client_sock)
                client_sock.close()
                backend_sock.close()
        except ConnectionError:
            self.sel.unregister(client_sock)
            client_sock.close()
            backend_sock.close()

# nutcracker.yml
redis-cluster:
  listen: 127.0.0.1:22121
  hash: fnv1a_64
  distribution: ketama
  timeout: 300
  preconnect: true
  server_retry_timeout: 30000
  server_failure_limit: 3
  servers:
    - 127.0.0.1:6379:1 server1
    - 127.0.0.1:6380:1 server2
    - 127.0.0.1:6381:1 server3

import threading
import queue

class ConnectionPool:
    """连接池管理"""

    def __init__(self, max_size=10, max_idle=5):
        self.max_size = max_size
        self.max_idle = max_idle
        self.pool = queue.Queue(maxsize=max_idle)
        self.active_count = 0
        self.lock = threading.Lock()
        self.backend_addr = None

    def connect(self, host, port):
        self.backend_addr = (host, port)
        # 预创建连接
        for _ in range(self.max_idle):
            conn = self._create_connection()
            self.pool.put(conn)

    def _create_connection(self):
        sock = socket.create_connection(self.backend_addr)
        with self.lock:
            self.active_count += 1
        return sock

    def get(self, timeout=5):
        """获取连接"""
        try:
            conn = self.pool.get_nowait()
            if self._is_alive(conn):
                return conn
            # 连接已失效，创建新的
            with self.lock:
                self.active_count -= 1
        except queue.Empty:
            pass

        with self.lock:
            if self.active_count < self.max_size:
                return self._create_connection()

        # 等待可用连接
        return self.pool.get(timeout=timeout)

    def put(self, conn):
        """归还连接"""
        try:
            self.pool.put_nowait(conn)
        except queue.Full:
            conn.close()
            with self.lock:
                self.active_count -= 1

    def _is_alive(self, conn):
        """检查连接是否存活"""
        try:
            conn.sendall(b"*1\r\n$4\r\nPING\r\n")
            resp = conn.recv(1024)
            return resp == b"+PONG\r\n"
        except:
            return False

# redis-py 连接池配置
import redis

pool = redis.ConnectionPool(
    host="127.0.0.1",
    port=6379,
    max_connections=50,         # 最大连接数
    socket_timeout=2.0,         # Socket 超时
    socket_connect_timeout=1.0, # 连接超时
    retry_on_timeout=True,      # 超时重试
    health_check_interval=30,   # 健康检查间隔
    decode_responses=True       # 自动解码
)

r = redis.Redis(connection_pool=pool)

def route_by_key(key, num_shards):
    """基于 key 的一致性哈希路由"""
    hash_val = crc16(key.encode())
    return hash_val % num_shards

READ_COMMANDS = {"GET", "MGET", "HGET", "HGETALL", "LRANGE", "SMEMBERS", "KEYS"}
WRITE_COMMANDS = {"SET", "MSET", "DEL", "HSET", "LPUSH", "SADD"}

def route_command(cmd, args, master_conn, slave_conn):
    """读写分离路由"""
    cmd_upper = cmd.upper()

    if cmd_upper in READ_COMMANDS:
        # 读命令发往从节点
        return slave_conn
    elif cmd_upper in WRITE_COMMANDS:
        # 写命令发往主节点
        return master_conn
    else:
        # 默认发往主节点
        return master_conn

class RESPProxy:
    def __init__(self):
        self.pending_requests = 0
        self.response_buffer = b""

    def forward_pipeline(self, client_sock, backend_sock, num_commands):
        """转发 Pipeline 请求"""
        # 读取所有命令
        commands = []
        for _ in range(num_commands):
            cmd = self._read_complete_command(client_sock)
            commands.append(cmd)

        # 一次性发送到后端
        backend_sock.sendall(b"".join(commands))

        # 收集响应
        responses = []
        for _ in range(num_commands):
            resp = self._read_response(backend_sock)
            responses.append(resp)

        # 返回客户端
        client_sock.sendall(b"".join(responses))

class PubSubProxy:
    def handle_subscribe(self, client_sock, backend_sock, channels):
        """处理订阅命令"""
        # 转发 SUBSCRIBE 命令
        cmd = encode_subscribe_command(channels)
        backend_sock.sendall(cmd)

        # 读取订阅确认
        for _ in range(len(channels)):
            resp = self._read_response(backend_sock)
            client_sock.sendall(self._encode_response(resp))

        # 进入消息转发模式
        self._forward_messages(client_sock, backend_sock)

    def _forward_messages(self, client_sock, backend_sock):
        """持续转发 Pub/Sub 消息"""
        while True:
            msg = self._read_response(backend_sock)
            client_sock.sendall(self._encode_response(msg))

def handle_error(self, client_sock, error_msg):
    """将错误转发给客户端"""
    error_resp = f"-{error_msg}\r\n".encode()
    client_sock.sendall(error_resp)

import concurrent.futures

class ThreadedProxy:
    def __init__(self, max_workers=100):
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)

    def handle_client(self, client_sock):
        """在独立线程中处理客户端"""
        backend_sock = self.connect_backend()

        while True:
            data = client_sock.recv(65536)
            if not data:
                break

            # 解析并路由
            commands = self.parse_commands(data)
            for cmd, args in commands:
                target = self.route(args)
                target.sendall(self.encode_command(cmd, args))
                resp = self.read_response(target)
                client_sock.sendall(resp)

        client_sock.close()
        backend_sock.close()

import asyncio

class AsyncProxy:
    def __init__(self):
        self.connections = {}

    async def handle_client(self, reader, writer):
        """异步处理客户端连接"""
        backend_reader, backend_writer = await asyncio.open_connection(
            *self.select_backend()
        )

        try:
            while True:
                data = await reader.read(65536)
                if not data:
                    break

                # 转发到后端
                backend_writer.write(data)
                await backend_writer.drain()

                # 读取响应
                resp = await backend_reader.read(65536)
                writer.write(resp)
                await writer.drain()
        finally:
            writer.close()
            backend_writer.close()

    async def start(self, host, port):
        server = await asyncio.start_server(
            self.handle_client, host, port
        )
        async with server:
            await server.serve_forever()

from prometheus_client import Counter, Histogram, Gauge

# 定义指标
REQUEST_COUNT = Counter('redis_proxy_requests_total', 'Total requests', ['command'])
REQUEST_LATENCY = Histogram('redis_proxy_request_duration_seconds', 'Request latency')
ACTIVE_CONNECTIONS = Gauge('redis_proxy_active_connections', 'Active connections')

def monitor_request(command, duration):
    REQUEST_COUNT.labels(command=command).inc()
    REQUEST_LATENCY.observe(duration)