$ telnet 127.0.0.1 6379
*3\r\n$9\r\nSUBSCRIBE\r\n$4\r\nnews\r\n$5\r\ncache\r\n

# 响应（每个频道一个确认）
*3\r\n$9\r\nsubscribe\r\n$4\r\nnews\r\n:1\r\n
*3\r\n$9\r\nsubscribe\r\n$5\r\ncache\r\n:2\r\n

import socket

s = socket.create_connection(("127.0.0.1", 6379))

# PUBLISH 本身是一个普通命令
cmd = "*3\r\n$7\r\nPUBLISH\r\n$4\r\nnews\r\n$5\r\nhello\r\n"
s.sendall(cmd.encode())
resp = s.recv(1024)
print(resp)  # b':0\r\n'（没有订阅者时返回 0）

import redis

class PubSubManager:
    def __init__(self, host="127.0.0.1", port=6379):
        # 独立的连接，不使用连接池
        self.redis = redis.Redis(host=host, port=port)
        self.pubsub = self.redis.pubsub()

    def subscribe(self, channel, callback):
        """订阅频道并注册回调"""
        self.pubsub.subscribe(**{channel: callback})
        # 启动独立线程监听消息
        thread = self.pubsub.run_in_thread(sleep_time=0.001)
        return thread

    def close(self):
        self.pubsub.close()
        self.redis.close()

# 使用
manager = PubSubManager()
def handle_message(message):
    if message["type"] == "message":
        print(f"Received: {message['data']}")

thread = manager.subscribe("news", handle_message)

import socket
import threading
from typing import Callable, Dict

class SimplePubSub:
    def __init__(self, host="127.0.0.1", port=6379):
        self.sock = socket.create_connection((host, port))
        self.buffer = b""
        self.handlers: Dict[str, Callable] = {}
        self.pattern_handlers: Dict[str, Callable] = {}
        self.running = False
        self._lock = threading.Lock()

    def _read_line(self) -> bytes:
        while b"\r\n" not in self.buffer:
            chunk = self.sock.recv(4096)
            if not chunk:
                raise ConnectionError("Connection closed")
            self.buffer += chunk
        pos = self.buffer.index(b"\r\n")
        line = self.buffer[:pos]
        self.buffer = self.buffer[pos + 2:]
        return line

    def _read_bulk_string(self) -> bytes | None:
        line = self._read_line()
        length = int(line[1:])
        if length == -1:
            return None
        while len(self.buffer) < length + 2:
            chunk = self.sock.recv(4096)
            if not chunk:
                raise ConnectionError("Connection closed")
            self.buffer += chunk
        data = self.buffer[:length]
        self.buffer = self.buffer[length + 2:]
        return data

    def _parse_response(self):
        line = self._read_line()
        prefix = chr(line[0])

        if prefix == ":":
            return int(line[1:])
        elif prefix == "$":
            length = int(line[1:])
            if length == -1:
                return None
            while len(self.buffer) < length + 2:
                chunk = self.sock.recv(4096)
                self.buffer += chunk
            data = self.buffer[:length]
            self.buffer = self.buffer[length + 2:]
            return data
        elif prefix == "*":
            count = int(line[1:])
            return [self._parse_response() for _ in range(count)]
        else:
            return line

    def subscribe(self, *channels: str):
        """订阅频道"""
        args = ["SUBSCRIBE"] + list(channels)
        parts = [f"*{len(args)}\r\n".encode()]
        for arg in args:
            arg_bytes = arg.encode()
            parts.append(f"${len(arg_bytes)}\r\n".encode())
            parts.append(arg_bytes)
            parts.append(b"\r\n")
        self.sock.sendall(b"".join(parts))

    def on(self, channel: str, callback: Callable):
        """注册消息处理器"""
        self.handlers[channel] = callback

    def listen(self):
        """监听消息（阻塞）"""
        self.running = True
        while self.running:
            msg = self._parse_response()
            if isinstance(msg, list) and len(msg) >= 3:
                msg_type = msg[0].decode() if isinstance(msg[0], bytes) else msg[0]

                if msg_type == "message":
                    channel = msg[1].decode()
                    data = msg[2].decode() if isinstance(msg[2], bytes) else msg[2]
                    if channel in self.handlers:
                        self.handlers[channel](channel, data)

                elif msg_type == "pmessage":
                    pattern = msg[1].decode()
                    channel = msg[2].decode()
                    data = msg[3].decode() if isinstance(msg[3], bytes) else msg[3]
                    if pattern in self.pattern_handlers:
                        self.pattern_handlers[pattern](pattern, channel, data)

    def close(self):
        self.running = False
        self.sock.close()


# 使用示例
def on_news(channel, message):
    print(f"[{channel}] {message}")

ps = SimplePubSub()
ps.on("news", on_news)
ps.subscribe("news", "alerts")
ps.listen()

def reconnect_and_resubscribe(pubsub, channels):
    """重连并重新订阅"""
    while True:
        try:
            pubsub.subscribe(*channels)
            pubsub.listen()
        except ConnectionError:
            time.sleep(1)
            # 重新连接并订阅
            pubsub.reconnect()