使用 Python 构建 Valkey Cluster 客户端：从零到生产实践指南

# 创建项目目录
mkdir valkey-glide-demo
cd valkey-glide-demo

# 安装 Valkey Glide 客户端
pip install valkey-glide

import asyncio
from typing import List, Tuple

from glide import (
    AllNodes,
    ClosingError,
    ConnectionError,
    GlideClusterClient,
    GlideClusterClientConfiguration,
    InfoSection,
    Logger,
    LogLevel,
    NodeAddress,
    RequestError,
    TimeoutError,
)


async def create_client(
    nodes_list: List[Tuple[str, int]] = [("valkey-demo.titbpn.clustercfg.use1.cache.amazonaws.com", 6379)]
) -> GlideClusterClient:
    """
    Creates and returns a GlideClusterClient instance.

    This function initializes a GlideClusterClient with the provided list of nodes.
    The nodes_list may contain the address of one or more cluster nodes, and the
    client will automatically discover all nodes in the cluster.

    Args:
        nodes_list (List[Tuple[str, int]]): A list of tuples where each tuple
            contains a host (str) and port (int). Defaults to [("localhost", 6379)].

    Returns:
        GlideClusterClient: An instance of GlideClusterClient connected to the discovered nodes.
    """
    addresses = [NodeAddress(host, port) for host, port in nodes_list]
    # Check `GlideClusterClientConfiguration` for additional options.
    config = GlideClusterClientConfiguration(
        addresses=addresses,
        client_name="test_cluster_client",
        # Enable this field if the servers are configured with TLS.
        # use_tls=True
    )
    return await GlideClusterClient.create(config)


async def app_logic(client: GlideClusterClient):
    """
    Executes the main logic of the application, performing basic operations
    such as SET, GET, PING, and INFO REPLICATION using the provided GlideClusterClient.

    Args:
        client (GlideClusterClient): An instance of GlideClusterClient.
    """
    # Send SET and GET
    set_response = await client.set("foo", "bar")
    Logger.log(LogLevel.INFO, "app", f"Set response is = {set_response!r}")

    get_response = await client.get("foo")
    assert isinstance(get_response, bytes)
    Logger.log(LogLevel.INFO, "app", f"Get response is = {get_response.decode()!r}")

    # Send PING to all primaries (according to Redis's PING request_policy)
    pong = await client.ping()
    Logger.log(LogLevel.INFO, "app", f"PING response is = {pong!r}")

    # Send INFO REPLICATION with routing option to all nodes
    info_repl_resps = await client.info([InfoSection.REPLICATION], AllNodes())
    Logger.log(
        LogLevel.INFO,
        "app",
        f"INFO REPLICATION responses from all nodes are=\n{info_repl_resps!r}",
    )


async def exec_app_logic():
    """
    Executes the application logic with exception handling.
    """
    while True:
        try:
            client = await create_client()
            return await app_logic(client)
        except asyncio.CancelledError:
            raise
        except ClosingError as e:
            # If the error message contains "NOAUTH", raise the exception
            # because it indicates a critical authentication issue.
            if "NOAUTH" in str(e):
                Logger.log(
                    LogLevel.ERROR,
                    "glide",
                    f"Authentication error encountered: {e}",
                )
            else:
                Logger.log(
                    LogLevel.WARN,
                    "glide",
                    f"Client has closed and needs to be re-created: {e}",
                )
            raise e
        except TimeoutError as e:
            # A request timed out. You may choose to retry the execution based on your application's logic
            Logger.log(LogLevel.ERROR, "glide", f"TimeoutError encountered: {e}")
            raise e
        except ConnectionError as e:
            # The client wasn't able to reestablish the connection within the given retries
            Logger.log(LogLevel.ERROR, "glide", f"ConnectionError encountered: {e}")
            raise e
        except RequestError as e:
            # Other error reported during a request, such as a server response error
            Logger.log(LogLevel.ERROR, "glide", f"RequestError encountered: {e}")
            raise e
        except Exception as e:
            Logger.log(LogLevel.ERROR, "glide", f"Unexpected error: {e}")
            raise e
        finally:
            try:
                await client.close()
            except Exception as e:
                Logger.log(
                    LogLevel.WARN,
                    "glide",
                    f"Error encountered while closing the client: {e}",
                )


def main():
    # In this example, we will utilize the client's logger for all log messages
    Logger.set_logger_config(LogLevel.INFO)
    # Optional - set the logger to write to a file
    # Logger.set_logger_config(LogLevel.INFO, file)
    asyncio.run(exec_app_logic())


if __name__ == "__main__":
    main()

Python
import os
import asyncio
import logging
import os
import zlib
from concurrent.futures import TimeoutError
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any
from typing import List, Tuple, Optional
from glide import GlideClusterClient, GlideClusterClientConfiguration, CoreCommands

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Get endpoint from environment variable
REDIS_ENDPOINT = os.getenv('REDIS_ENDPOINT', 'valkey-demo.titbpn.clustercfg.use1.cache.amazonaws.com')
REDIS_PORT = int(os.getenv('REDIS_PORT', '6379'))

# Define channels and patterns as constants
EXACT_CHANNELS = {"ch1", "ch2"}
PATTERN_CHANNELS = {"chat*"}


@dataclass
class NodeAddress:
    host: str = "valkey-demo.titbpn.clustercfg.use1.cache.amazonaws.com"
    port: int = 6379


class GlideClientManager:
    _instance = None

    def __init__(self):
        self.client = None

    @classmethod
    async def get_instance(cls, callback, context=None):
        if cls._instance is None:
            cls._instance = cls()
            cls._instance.client = await initialize_clients(callback, context)
        return cls._instance.client


# Create a single node address configuration to be reused
node_config = [NodeAddress(REDIS_ENDPOINT, REDIS_PORT)]


def callback(msg: CoreCommands.PubSubMsg, context: Any):
    logger.info(f"Received {msg}, context {context}")


async def initialize_clients(callback, context: Optional[object] = None):
    # Combined configuration for both listening and publishing
    client_config = GlideClusterClientConfiguration(
        node_config,
        pubsub_subscriptions=GlideClusterClientConfiguration.PubSubSubscriptions(
            channels_and_patterns={
                GlideClusterClientConfiguration.PubSubChannelModes.Exact: EXACT_CHANNELS,
                GlideClusterClientConfiguration.PubSubChannelModes.Pattern: PATTERN_CHANNELS
            },
            callback=callback,
            context=context,
        )
    )

    try:
        # Create a single client for both listening and publishing
        client = await GlideClusterClient.create(client_config)
        return client
    except Exception as e:
        print(f"Failed to initialize client: {e}")
        raise


@asynccontextmanager
async def get_client(callback, context=None):
    client = None
    try:
        client = await initialize_clients(callback, context)
        yield client
    finally:
        if client:
            await client.close()


async def publish_message(client, message: str, channel: str):
    try:
        await client.publish(message, channel)
    except Exception as e:
        print(f"Failed to publish message: {e}")
        raise


async def publish_compressed_message(client, message: str, channel: str):
    compressed_message = zlib.compress(message.encode())
    await client.publish(compressed_message, channel)


@dataclass
class PublishResult:
    success: bool
    message: str
    channel: str
    error: Optional[Exception] = None


async def publish_messages_batch(
        client: GlideClusterClient,
        messages: List[Tuple[str, str]],
        chunk_size: int = 100,
        timeout: Optional[float] = 30.0
) -> List[PublishResult]:
    """
    Publish multiple messages in batches with improved error handling and performance

    Args:
        client: GlideClusterClient instance
        messages: List of (message, channel) tuples
        chunk_size: Number of messages to process in each batch
        timeout: Maximum time in seconds to wait for batch completion

    Returns:
        List of PublishResult objects containing publish status and details
    """
    results: List[PublishResult] = []

    async def publish_single(message: str, channel: str) -> PublishResult:
        try:
            await client.publish(message, channel)
            return PublishResult(success=True, message=message, channel=channel)
        except Exception as e:
            logger.error(f"Failed to publish message to channel {channel}: {str(e)}")
            return PublishResult(success=False, message=message, channel=channel, error=e)

    # Process messages in chunks
    for i in range(0, len(messages), chunk_size):
        chunk = messages[i:i + chunk_size]
        try:
            chunk_results = await asyncio.wait_for(
                asyncio.gather(*[
                    publish_single(message, channel)
                    for message, channel in chunk
                ]),
                timeout=timeout
            )
            results.extend(chunk_results)

        except TimeoutError as e:
            results.extend([
                PublishResult(False, message, channel, e)
                for message, channel in chunk
            ])

    return results


async def main():
    async with get_client(callback) as client:
        try:
            # Example of batch publishing
            messages = [
                ("Test message 1", "ch1"),
                ("Test message 2", "ch2")
            ]
            await publish_messages_batch(client, messages)

            # Your work/wait logic here
            await asyncio.sleep(1)  # Example delay

        except Exception as e:
            logger.error(f"Error in main: {e}")


if __name__ == "__main__":
    asyncio.run(main())