Python脚本如何生成Pulsar配置

本文目录导读：

1. 基本Pulsar配置生成
2. 配置文件持久化
3. 高级配置生成
4. 配置验证工具
5. 命令行工具示例

我来详细介绍如何使用Python生成Apache Pulsar的配置。

基本Pulsar配置生成

基础配置生成器

import yaml
import json
from typing import Dict, Any, Optional
from pathlib import Path
class PulsarConfigGenerator:
    """Pulsar配置生成器"""
    def __init__(self):
        self.config = {}
    def generate_broker_config(self, **kwargs) -> Dict[str, Any]:
        """生成Broker配置"""
        config = {
            # 基础配置
            "brokerServicePort": kwargs.get("port", 6650),
            "webServicePort": kwargs.get("web_port", 8080),
            "clusterName": kwargs.get("cluster_name", "standalone"),
            "zookeeperServers": kwargs.get("zk_servers", "localhost:2181"),
            "configurationStoreServers": kwargs.get("config_store", "localhost:2181"),
            # 存储配置
            "managedLedgerDefaultAckQuorum": kwargs.get("ack_quorum", 2),
            "managedLedgerDefaultEnsembleSize": kwargs.get("ensemble_size", 3),
            "managedLedgerDefaultWriteQuorum": kwargs.get("write_quorum", 3),
            "managedLedgerMaxEntriesPerLedger": kwargs.get("max_entries", 50000),
            # 认证配置
            "authenticationEnabled": kwargs.get("auth_enabled", False),
            "authenticationProviders": kwargs.get("auth_providers", []),
            "authorizationEnabled": kwargs.get("authz_enabled", False),
            # 租户和命名空间配置
            "defaultTenant": kwargs.get("default_tenant", "public"),
            "defaultNamespace": kwargs.get("default_namespace", "default"),
            # 消息配置
            "maxMessageSize": kwargs.get("max_msg_size", 5242880),  # 5MB
            "maxConcurrentLookupRequest": kwargs.get("max_lookup", 5000),
            # 其他性能配置
            "numIOThreads": kwargs.get("io_threads", 16),
            "numWorkerThreads": kwargs.get("worker_threads", 20),
        }
        # 自定义配置
        if kwargs.get("custom_config"):
            config.update(kwargs["custom_config"])
        return config
    def generate_client_config(self, **kwargs) -> Dict[str, Any]:
        """生成客户端配置"""
        config = {
            "serviceUrl": kwargs.get("service_url", "pulsar://localhost:6650"),
            "listenerName": kwargs.get("listener_name", None),
            "operationTimeout": kwargs.get("timeout", 30),
            "ioThreads": kwargs.get("io_threads", 4),
            "newProxyArgs": kwargs.get("proxy_args", {}),
            # TLS配置
            "useTls": kwargs.get("use_tls", False),
            "tlsTrustCertsFilePath": kwargs.get("tls_cert_path", ""),
            "tlsAllowInsecureConnection": kwargs.get("tls_insecure", False),
            # 认证配置
            "authPluginClassName": kwargs.get("auth_plugin", ""),
            "authParamsString": kwargs.get("auth_params", ""),
        }
        return config
    def generate_producer_config(self, **kwargs) -> Dict[str, Any]:
        """生成生产者配置"""
        config = {
            "topicName": kwargs.get("topic"),
            "producerName": kwargs.get("producer_name", ""),
            "sendTimeoutMs": kwargs.get("send_timeout", 30000),
            "blockIfQueueFull": kwargs.get("block_queue", True),
            "maxPendingMessages": kwargs.get("max_pending", 1000),
            "maxPendingMessagesAcrossPartitions": kwargs.get("max_pending_total", 50000),
            "batchingEnabled": kwargs.get("batching", True),
            "batchingMaxMessages": kwargs.get("batch_size", 1000),
            "batchingMaxPublishDelayMs": kwargs.get("batch_delay", 10),
            "compressionType": kwargs.get("compression", "NONE"),
        }
        return config
    def generate_consumer_config(self, **kwargs) -> Dict[str, Any]:
        """生成消费者配置"""
        config = {
            "topicNames": kwargs.get("topics", []),
            "subscriptionName": kwargs.get("subscription"),
            "subscriptionType": kwargs.get("sub_type", "Exclusive"),
            "subscriptionInitialPosition": kwargs.get("initial_position", "Latest"),
            "receiverQueueSize": kwargs.get("queue_size", 1000),
            "maxTotalReceiverQueueSizeAcrossPartitions": kwargs.get("max_queue_total", 50000),
            "ackTimeoutMs": kwargs.get("ack_timeout", 0),
            "negativeAckRedeliveryDelayMs": kwargs.get("nack_delay", 60000),
            "consumerName": kwargs.get("consumer_name", ""),
        }
        return config
# 使用示例
if __name__ == "__main__":
    generator = PulsarConfigGenerator()
    # 生成Broker配置
    broker_config = generator.generate_broker_config(
        port=6650,
        web_port=8080,
        cluster_name="production-cluster",
        zk_servers="zk1:2181,zk2:2181,zk3:2181",
        auth_enabled=True,
        auth_providers=["org.apache.pulsar.broker.authentication.AuthenticationProviderToken"],
        custom_config={
            "allowAutoTopicCreation": True,
            "allowAutoTopicCreationType": "partitioned"
        }
    )
    print("Broker Config:", json.dumps(broker_config, indent=2))

配置文件持久化

class ConfigPersister:
    """配置持久化工具"""
    @staticmethod
    def save_as_yaml(config: Dict, filepath: str):
        """保存为YAML格式"""
        with open(filepath, 'w', encoding='utf-8') as f:
            yaml.dump(config, f, default_flow_style=False, allow_unicode=True)
    @staticmethod
    def save_as_properties(config: Dict, filepath: str):
        """保存为Properties格式（Pulsar常用）"""
        with open(filepath, 'w', encoding='utf-8') as f:
            for key, value in config.items():
                if isinstance(value, list):
                    value = ','.join(str(v) for v in value)
                f.write(f"{key}={value}\n")
    @staticmethod
    def save_as_json(config: Dict, filepath: str):
        """保存为JSON格式"""
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
    @staticmethod
    def load_config(filepath: str) -> Dict:
        """加载配置文件"""
        path = Path(filepath)
        ext = path.suffix.lower()
        with open(filepath, 'r', encoding='utf-8') as f:
            if ext == '.yaml' or ext == '.yml':
                return yaml.safe_load(f)
            elif ext == '.json':
                return json.load(f)
            elif ext == '.properties' or ext == '.conf':
                config = {}
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#'):
                        key, value = line.split('=', 1)
                        config[key.strip()] = value.strip()
                return config
            else:
                raise ValueError(f"Unsupported file format: {ext}")
# 使用示例
def save_configurations():
    generator = PulsarConfigGenerator()
    persister = ConfigPersister()
    # 生成并保存各种配置
    broker_config = generator.generate_broker_config(cluster_name="prod")
    # 保存为Pulsar原生格式
    persister.save_as_properties(broker_config, "conf/broker.conf")
    # 保存为YAML格式（方便阅读）
    persister.save_as_yaml(broker_config, "conf/broker_config.yaml")

高级配置生成

class AdvancedPulsarConfig:
    """高级配置生成"""
    @staticmethod
    def generate_tls_config(**kwargs) -> Dict:
        """生成TLS配置"""
        return {
            "tlsEnabled": True,
            "tlsCertificateFilePath": kwargs.get("cert_path"),
            "tlsKeyFilePath": kwargs.get("key_path"),
            "tlsTrustCertsFilePath": kwargs.get("trust_cert_path"),
            "tlsAllowInsecureConnection": kwargs.get("allow_insecure", False),
            "tlsRequireTrustedClientCertOnConnect": kwargs.get("require_client_cert", False),
        }
    @staticmethod
    def generate_authentication_config(auth_type: str, **kwargs) -> Dict:
        """生成认证配置"""
        if auth_type == "token":
            return {
                "authenticationEnabled": True,
                "authenticationProviders": [
                    "org.apache.pulsar.broker.authentication.AuthenticationProviderToken"
                ],
                "tokenSecretKey": kwargs.get("secret_key", ""),
                "tokenPublicKey": kwargs.get("public_key", ""),
            }
        elif auth_type == "oauth2":
            return {
                "authenticationEnabled": True,
                "authenticationProviders": [
                    "org.apache.pulsar.broker.authentication.AuthenticationProviderToken"
                ],
                "oauth2IssuerUrl": kwargs.get("issuer_url"),
                "oauth2Audience": kwargs.get("audience"),
            }
        elif auth_type == "athenz":
            return {
                "authenticationEnabled": True,
                "authenticationProviders": [
                    "org.apache.pulsar.broker.authentication.AuthenticationProviderAthenz"
                ],
                "athenzDomainNames": kwargs.get("domain_name"),
                "athenzTenantDomains": kwargs.get("tenant_domain"),
            }
        else:
            raise ValueError(f"Unsupported auth type: {auth_type}")
    @staticmethod
    def generate_bookkeeper_config(**kwargs) -> Dict:
        """生成BookKeeper配置"""
        return {
            "bookkeeperClient": {
                "ensembleSize": kwargs.get("ensemble_size", 3),
                "writeQuorumSize": kwargs.get("write_quorum", 3),
                "ackQuorumSize": kwargs.get("ack_quorum", 2),
                "metadataServiceUri": kwargs.get("metadata_uri", "zk://localhost:2181/ledgers"),
            },
            "bookie": {
                "journalDirectory": kwargs.get("journal_dir", "/data/bookkeeper/journal"),
                "ledgerDirectories": kwargs.get("ledger_dirs", ["/data/bookkeeper/ledger"]),
                "indexDirectories": kwargs.get("index_dirs", ["/data/bookkeeper/index"]),
                "journalMaxSizeMB": kwargs.get("journal_max_size", 2048),
                "journalPageCacheFlushIntervalMS": 1000,
            }
        }
# 使用示例
def generate_complete_config():
    """生成完整的Pulsar配置"""
    config = {
        "broker": PulsarConfigGenerator().generate_broker_config(
            cluster_name="production",
            zk_servers="zk1:2181,zk2:2181,zk3:2181"
        ),
        "tls": AdvancedPulsarConfig.generate_tls_config(
            cert_path="/etc/pulsar/tls/broker.crt",
            key_path="/etc/pulsar/tls/broker.key"
        ),
        "authentication": AdvancedPulsarConfig.generate_authentication_config(
            "token",
            secret_key="/etc/pulsar/token/secret.key"
        ),
        "bookkeeper": AdvancedPulsarConfig.generate_bookkeeper_config(
            ensemble_size=3,
            write_quorum=3,
            ack_quorum=2
        )
    }
    return config

配置验证工具

class ConfigValidator:
    """配置验证器"""
    @staticmethod
    def validate_broker_config(config: Dict) -> bool:
        """验证Broker配置"""
        required_fields = ["clusterName", "zookeeperServers"]
        errors = []
        for field in required_fields:
            if field not in config:
                errors.append(f"Missing required field: {field}")
        # 验证端口范围
        if "brokerServicePort" in config:
            if not (1024 <= config["brokerServicePort"] <= 65535):
                errors.append("Invalid broker port range")
        # 验证整数配置
        for field in ["numIOThreads", "numWorkerThreads"]:
            if field in config and config[field] < 1:
                errors.append(f"{field} must be positive")
        if errors:
            print("Configuration errors:", errors)
            return False
        return True
    @staticmethod
    def validate_client_config(config: Dict) -> bool:
        """验证客户端配置"""
        if "serviceUrl" not in config:
            print("Missing service URL")
            return False
        # 验证URL格式
        url = config["serviceUrl"]
        if not url.startswith("pulsar://") and not url.startswith("pulsar+ssl://"):
            print("Invalid service URL format")
            return False
        return True
# 使用示例
def validate_and_save():
    generator = PulsarConfigGenerator()
    validator = ConfigValidator()
    broker_config = generator.generate_broker_config(
        cluster_name="test",
        zk_servers="localhost:2181"
    )
    if validator.validate_broker_config(broker_config):
        ConfigPersister.save_as_properties(broker_config, "valid_broker.conf")
        print("Configuration saved successfully")
    else:
        print("Configuration validation failed")

命令行工具示例

import argparse
def create_cli():
    """创建命令行接口"""
    parser = argparse.ArgumentParser(description="Generate Pulsar configuration")
    parser.add_argument("--type", choices=["broker", "client", "producer", "consumer"],
                       required=True, help="Configuration type")
    parser.add_argument("--output", help="Output file path")
    parser.add_argument("--format", choices=["properties", "yaml", "json"],
                       default="properties", help="Output format")
    parser.add_argument("--cluster", default="standalone", help="Cluster name")
    parser.add_argument("--zk-servers", default="localhost:2181", help="ZooKeeper servers")
    parser.add_argument("--port", type=int, default=6650, help="Broker port")
    return parser
if __name__ == "__main__":
    parser = create_cli()
    args = parser.parse_args()
    generator = PulsarConfigGenerator()
    persister = ConfigPersister()
    if args.type == "broker":
        config = generator.generate_broker_config(
            cluster_name=args.cluster,
            zk_servers=args.zk_servers,
            port=args.port
        )
    # 其他类型的处理...
    if args.output:
        if args.format == "properties":
            persister.save_as_properties(config, args.output)
        elif args.format == "yaml":
            persister.save_as_yaml(config, args.output)
        else:
            persister.save_as_json(config, args.output)

这些工具可以帮助你：

1. 动态生成Pulsar配置
2. 支持多种输出格式（Properties、YAML、JSON）
3. 包含配置验证功能
4. 支持TLS和认证配置
5. 提供命令行接口

根据你的具体需求,可以扩展这些工具来支持更多的配置项和验证规则。