Python脚本如何生成EMQX配置

wen 实用脚本 2026-07-05 80

本文目录导读：

Python脚本如何生成EMQX配置

基础配置生成
高级配置生成
配置文件验证
完整的配置管理工具
使用说明

我来详细介绍如何使用Python生成EMQX配置，EMQX主要使用HOCON格式的配置文件,我将提供几种常见的生成方法。

基础配置生成

1 使用字符串模板

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
def generate_basic_config(broker_name="emqx@127.0.0.1", 
                          listener_port=1883,
                          max_connections=1000000):
    """
    生成基本的EMQX配置
    """
    config = f"""
# EMQX Basic Configuration
node.name = {broker_name}
# MQTT/TCP Listener
listeners.tcp.default {{
    bind = "0.0.0.0:{listener_port}"
    max_connections = {max_connections}
}}
# Dashboard
dashboard {{
    listeners.http {{
        bind = "0.0.0.0:18083"
    }}
}}
# MQTT Settings
mqtt {{
    max_packet_size = "1MB"
    max_clientid_len = 128
    max_topic_depth = 128
}}
"""
    return config
# 生成配置
config_content = generate_basic_config()
with open('emqx.conf', 'w') as f:
    f.write(config_content)
print("基础配置文件已生成")

2 使用字典结构生成HOCON格式

import json
from typing import Dict, Any, List
class EMQXConfigGenerator:
    """EMQX配置生成器"""
    def __init__(self):
        self.config = {}
    def set_node_config(self, name="emqx@127.0.0.1", 
                        cookie="emqxsecretcookie",
                        role="core"):
        self.config['node'] = {
            'name': name,
            'cookie': cookie,
            'role': role
        }
        return self
    def add_tcp_listener(self, port=1883, max_connections=1024000):
        self.config.setdefault('listeners', {})
        self.config['listeners'].setdefault('tcp', {})
        self.config['listeners']['tcp']['default'] = {
            'bind': f"0.0.0.0:{port}",
            'max_connections': max_connections
        }
        return self
    def add_ssl_listener(self, port=8883, cert_file=None, key_file=None):
        self.config.setdefault('listeners', {})
        self.config['listeners'].setdefault('ssl', {})
        ssl_config = {
            'bind': f"0.0.0.0:{port}",
            'ssl_options': {}
        }
        if cert_file:
            ssl_config['ssl_options']['certfile'] = cert_file
        if key_file:
            ssl_config['ssl_options']['keyfile'] = key_file
        self.config['listeners']['ssl']['default'] = ssl_config
        return self
    def add_authentication(self, auth_type='built-in-database', 
                           users: List[Dict] = None):
        if not self.config.get('authentication'):
            self.config['authentication'] = []
        auth_config = {
            'mechanism': 'password_based',
            'backend': auth_type
        }
        if users:
            auth_config['users'] = users
        self.config['authentication'].append(auth_config)
        return self
    def generate_hocon(self, obj=None, indent=0):
        """将配置字典转换为HOCON格式字符串"""
        if obj is None:
            obj = self.config
        lines = []
        prefix = " " * indent
        if isinstance(obj, dict):
            first = True
            for key, value in obj.items():
                if isinstance(value, (dict, list)):
                    if first:
                        if indent > 0:
                            lines.append(f"{prefix}{key} {{")
                        else:
                            lines.append(f"{prefix}{key} {{")
                    else:
                        lines[-1] += f"\n{prefix}{key} {{"
                    if isinstance(value, dict):
                        lines.append(self.generate_hocon(value, indent + 4))
                    elif isinstance(value, list):
                        for item in value:
                            if isinstance(item, dict):
                                lines.append(self.generate_hocon(item, indent + 4))
                            else:
                                lines.append(f"{prefix}    - {item}")
                    if lines[-1].strip().endswith('{'):
                        lines[-1] += "}"
                    else:
                        lines.append(f"{prefix}}}")
                else:
                    if isinstance(value, str):
                        if value.startswith('0.0.0.0:'):
                            lines.append(f"{prefix}{key} = \"{value}\"")
                        else:
                            lines.append(f"{prefix}{key} = {value}")
                    elif isinstance(value, bool):
                        lines.append(f"{prefix}{key} = {'true' if value else 'false'}")
                    else:
                        lines.append(f"{prefix}{key} = {value}")
                first = False
        elif isinstance(obj, list):
            for item in obj:
                lines.append(f"{prefix}- {item}")
        else:
            lines.append(f"{prefix}{obj}")
        return '\n'.join(lines)
    def save_to_file(self, filename='emqx.conf'):
        """保存配置到文件"""
        config_str = self.generate_hocon()
        with open(filename, 'w') as f:
            f.write(config_str)
        print(f"配置文件已保存到: {filename}")
# 使用示例
generator = EMQXConfigGenerator()
generator.set_node_config("emqx@192.168.1.100", cookie="mysecret")
generator.add_tcp_listener(1883, 100000)
generator.add_ssl_listener(8883, 
                          cert_file="/etc/emqx/certs/cert.pem",
                          key_file="/etc/emqx/certs/key.pem")
generator.add_authentication('built-in-database', [
    {'username': 'admin', 'password': 'password123'},
    {'username': 'user1', 'password': 'userpass123'}
])
generator.save_to_file('emqx.conf')

高级配置生成

1 从YAML模板生成

import yaml
import json
from pathlib import Path
class EMQXConfigFromTemplate:
    """从YAML模板生成EMQX配置"""
    def __init__(self, template_file=None):
        self.template = {}
        if template_file and Path(template_file).exists():
            with open(template_file, 'r') as f:
                self.template = yaml.safe_load(f)
    def create_template(self) -> dict:
        """创建配置模板"""
        return {
            'node': {
                'name': 'emqx@127.0.0.1',
                'cookie': 'emqxsecretcookie',
                'role': 'core'
            },
            'listeners': {
                'tcp': {
                    'default': {
                        'bind': '0.0.0.0:1883',
                        'max_connections': 1024000
                    }
                }
            },
            'mqtt': {
                'max_packet_size': '1MB',
                'max_clientid_len': 128,
                'retry_interval': '30s'
            },
            'authentication': [
                {
                    'mechanism': 'password_based',
                    'backend': 'built-in-database',
                    'enable': True,
                    'user_id_type': 'username',
                    'password_hash_algorithm': {
                        'name': 'sha256',
                        'salt_position': 'suffix'
                    }
                }
            ],
            'authorization': {
                'no_match': 'deny',
                'sources': [
                    {
                        'type': 'built-in-database',
                        'enable': True
                    }
                ]
            },
            'broker': {
                'sysmsg_interval': '30s',
                'session_expiry_interval': '2h'
            }
        }
    def generate_from_template(self, outputs: dict = None) -> str:
        """
        根据模板生成配置
        Args:
            outputs: 覆盖模板的输出配置
        Returns:
            HOCON格式的配置字符串
        """
        config = self.create_template()
        if outputs:
            self._deep_update(config, outputs)
        return self._dict_to_hocon(config)
    def _deep_update(self, base: dict, updates: dict) -> dict:
        """递归更新字典"""
        for key, value in updates.items():
            if key in base and isinstance(base[key], dict) and isinstance(value, dict):
                self._deep_update(base[key], value)
            else:
                base[key] = value
        return base
    def _dict_to_hocon(self, obj: dict, indent: int = 0) -> str:
        """将字典转换为HOCON格式"""
        lines = []
        prefix = " " * indent
        for key, value in obj.items():
            if isinstance(value, dict):
                lines.append(f"{prefix}{key} {{")
                lines.append(self._dict_to_hocon(value, indent + 4))
                lines.append(f"{prefix}}}")
            elif isinstance(value, list):
                for item in value:
                    if isinstance(item, dict):
                        lines.append(f"{prefix}{key} = [")
                        lines.append(self._dict_to_hocon(item, indent + 4))
                        lines.append(f"{prefix}]")
                    else:
                        lines.append(f"{prefix}{key} = [{value}]")
            else:
                if isinstance(value, bool):
                    lines.append(f"{prefix}{key} = {'true' if value else 'false'}")
                elif isinstance(value, (int, float)):
                    lines.append(f"{prefix}{key} = {value}")
                else:
                    lines.append(f"{prefix}{key} = \"{value}\"")
        return "\n".join(lines)
# 使用示例
generator = EMQXConfigFromTemplate()
output_overrides = {
    'node': {'name': 'emqx@10.0.0.1'},
    'listeners': {
        'tcp': {
            'default': {
                'bind': '0.0.0.0:1883',
                'max_connections': 500000
            }
        },
        'ssl': {
            'default': {
                'bind': '0.0.0.0:8883',
                'ssl_options': {
                    'certfile': '/etc/emqx/certs/cert.pem',
                    'keyfile': '/etc/emqx/certs/key.pem'
                }
            }
        }
    }
}
config_str = generator.generate_from_template(output_overrides)
with open('emqx_custom.conf', 'w') as f:
    f.write(config_str)
print("自定义配置已生成")

2 集群配置生成

import json
from typing import List
class EMQXClusterConfigGenerator:
    """EMQX集群配置生成器"""
    def __init__(self):
        self.nodes = []
        self.cluster_config = {}
    def add_node(self, node_name: str, ip: str, 
                 discovery_method: str = 'static',
                 ports: dict = None) -> 'EMQXClusterConfigGenerator':
        """
        添加集群节点
        Args:
            node_name: 节点名称
            ip: 节点IP地址
            discovery_method: 发现方式
            ports: 端口配置
        """
        if ports is None:
            ports = {'mqtt': 1883, 'mqtts': 8883, 'ws': 8083, 'wss': 8084}
        node_config = {
            'node': {
                'name': f"{node_name}@{ip}",
                'role': 'core'
            },
            'listeners': {
                'tcp': {
                    'default': {
                        'bind': f"0.0.0.0:{ports['mqtt']}",
                        'max_connections': 1024000
                    }
                }
            },
            'cluster': {
                'discovery_strategy': discovery_method,
                'static': {
                    'seeds': []  # 将在后续填充
                }
            }
        }
        self.nodes.append({
            'name': node_name,
            'ip': ip,
            'config': node_config
        })
        return self
    def build_cluster_config(self) -> str:
        """
        构建完整的集群配置
        """
        # 填充所有节点的种子列表
        all_seeds = []
        for node in self.nodes:
            seed = f"{node['name']}@{node['ip']}:4370"
            all_seeds.append(seed)
        # 为每个节点生成配置文件
        configs = []
        for node in self.nodes:
            config = node['config'].copy()
            config['cluster']['static']['seeds'] = all_seeds
            # 添加集群相关配置
            config.update({
                'cluster': {
                    'name': 'emqx-cluster',
                    'discovery_strategy': 'static',
                    'static': {
                        'seeds': all_seeds
                    },
                    'core_nodes': [n['name'] for n in self.nodes],
                    'replicant_nodes': []
                },
                'node': {
                    'role': 'core',
                    'data_dir': f"data/{node['name']}"
                }
            })
            configs.append({
                'node_name': node['name'],
                'config': config
            })
        # 为每个节点保存配置文件
        for idx, node_config in enumerate(configs):
            filename = f"emqx_{node_config['node_name']}.conf"
            with open(filename, 'w') as f:
                f.write(self._format_config(node_config['config']))
            print(f"节点 {node_config['node_name']} 配置已保存到 {filename}")
        return "集群配置生成完成"
    def _format_config(self, config: dict, indent: int = 0) -> str:
        """格式化配置为HOCON"""
        result = []
        prefix = " " * indent
        for key, value in config.items():
            if value is None:
                continue
            if isinstance(value, dict):
                result.append(f"{prefix}{key} {{")
                result.append(self._format_config(value, indent + 4))
                result.append(f"{prefix}}}")
            elif isinstance(value, list):
                if all(isinstance(item, dict) for item in value):
                    result.append(f"{prefix}{key} = [")
                    for item in value:
                        result.append(f"{prefix}    {{")
                        result.append(self._format_config(item, indent + 8))
                        result.append(f"{prefix}    }}")
                    result.append(f"{prefix}]")
                else:
                    items = ', '.join([f'"{i}"' if isinstance(i, str) else str(i) for i in value])
                    result.append(f"{prefix}{key} = [{items}]")
            else:
                if isinstance(value, bool):
                    result.append(f"{prefix}{key} = {'true' if value else 'false'}")
                elif isinstance(value, (int, float)):
                    result.append(f"{prefix}{key} = {value}")
                else:
                    result.append(f"{prefix}{key} = \"{value}\"")
        return "\n".join(result)
# 使用示例
cluster_gen = EMQXClusterConfigGenerator()
# 添加3个节点
cluster_gen.add_node("emqx1", "192.168.1.10")
cluster_gen.add_node("emqx2", "192.168.1.11")
cluster_gen.add_node("emqx3", "192.168.1.12")
# 生成集群配置
result = cluster_gen.build_cluster_config()
print(result)

配置文件验证

import re
from typing import Tuple, List
class EMQXConfigValidator:
    """EMQX配置验证器"""
    @staticmethod
    def validate_hocon_syntax(config_str: str) -> Tuple[bool, List[str]]:
        """
        验证HOCON语法
        Returns: (是否有效, 错误列表)
        """
        errors = []
        # 检查括号匹配
        brace_stack = []
        bracket_stack = []
        for i, char in enumerate(config_str):
            if char == '{':
                brace_stack.append(i)
            elif char == '}':
                if not brace_stack:
                    errors.append(f"行 {config_str[:i].count(chr(10)) + 1}: 多余的 }")
                else:
                    brace_stack.pop()
            elif char == '[':
                bracket_stack.append(i)
            elif char == ']':
                if not bracket_stack:
                    errors.append(f"行 {config_str[:i].count(chr(10)) + 1}: 多余的 ]")
                else:
                    bracket_stack.pop()
        if brace_stack:
            errors.append(f"缺少 {len(brace_stack)} 个 }")
        if bracket_stack:
            errors.append(f"缺少 {len(bracket_stack)} 个 ]")
        # 检查配置项格式
        lines = config_str.split('\n')
        for i, line in enumerate(lines, 1):
            stripped = line.strip()
            if stripped and not stripped.startswith('#') and not stripped.startswith('//'):
                # 检查是否有基本的键值对格式
                if '=' in stripped and not stripped.endswith('{') and not stripped.endswith('}'):
                    parts = stripped.split('=')
                    if len(parts) != 2:
                        errors.append(f"行 {i}: 无效的配置格式: {stripped}")
        return len(errors) == 0, errors
    @staticmethod
    def validate_node_config(config: dict) -> List[str]:
        """
        验证节点配置
        """
        errors = []
        node = config.get('node', {})
        if not node:
            errors.append("缺少 node 配置")
        else:
            if 'name' not in node:
                errors.append("缺少 node.name")
            else:
                name = node['name']
                if '@' not in name:
                    errors.append(f"node.name '{name}' 格式无效，应为 name@ip")
        listeners = config.get('listeners', {})
        if not listeners:
            errors.append("缺少 listeners 配置")
        else:
            tcp = listeners.get('tcp', {})
            ssl = listeners.get('ssl', {})
            if not tcp and not ssl:
                errors.append("至少需要一个监听器 (tcp 或 ssl)")
        if not config.get('authentication'):
            errors.append("缺少 authentication 配置")
        return errors
    @staticmethod
    def validate_config_file(filepath: str) -> Tuple[bool, List[str]]:
        """
        验证配置文件
        """
        try:
            with open(filepath, 'r') as f:
                content = f.read()
            all_errors = []
            # 语法检查
            syntax_valid, syntax_errors = EMQXConfigValidator.validate_hocon_syntax(content)
            all_errors.extend(syntax_errors)
            if syntax_valid:
                # 内容检查
                config = {}
                exec(f"config = {content}", globals(), {'config': config})
                config_errors = EMQXConfigValidator.validate_node_config(config)
                all_errors.extend(config_errors)
            return len(all_errors) == 0, all_errors
        except Exception as e:
            return False, [f"读取配置文件失败: {str(e)}"]
# 使用验证器
validator = EMQXConfigValidator()
is_valid, errors = validator.validate_config_file('emqx.conf')
if is_valid:
    print("配置验证通过")
else:
    print("配置验证失败:")
    for error in errors:
        print(f"  - {error}")

完整的配置管理工具

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import sys
import json
import argparse
from pathlib import Path
from typing import Dict, Any
from datetime import datetime
class EMQXConfigManager:
    """EMQX配置管理器"""
    def __init__(self, config_dir: str = "/etc/emqx"):
        self.config_dir = Path(config_dir)
        self.configs = {}
        self.backup_dir = self.config_dir / "backups"
    def create_default_config(self) -> str:
        """创建默认配置"""
        config = {
            'node': {
                'name': 'emqx@127.0.0.1',
                'cookie': 'emqxsecretcookie',
                'role': 'core'
            },
            'listeners': {
                'tcp': {
                    'default': {
                        'bind': '0.0.0.0:1883',
                        'max_connections': 1024000
                    }
                },
                'ssl': {
                    'default': {
                        'bind': '0.0.0.0:8883'
                    }
                }
            },
            'mqtt': {
                'max_packet_size': '1MB',
                'max_clientid_len': 128,
                'retry_interval': '30s',
                'session_expiry_interval': '2h'
            },
            'dashboard': {
                'listeners': {
                    'http': {
                        'bind': '0.0.0.0:18083'
                    }
                },
                'default_password': 'public'
            },
            'authentication': [
                {
                    'mechanism': 'password_based',
                    'backend': 'built-in-database',
                    'enable': True
                }
            ],
            'authorization': {
                'no_match': 'deny',
                'sources': [
                    {
                        'type': 'built-in-database',
                        'enable': True
                    }
                ]
            },
            'broker': {
                'sysmsg_interval': '30s'
            },
            'log': {
                'console_handler': {
                    'enable': True,
                    'level': 'info'
                },
                'file_handlers': {
                    'default': {
                        'enable': True,
                        'level': 'warning',
                        'file': '${EMQX_LOG_DIR}/emqx.log'
                    }
                }
            },
            'stats': {
                'enable': True
            }
        }
        return self._dict_to_hocon(config)
    def _dict_to_hocon(self, obj: dict, indent: int = 0) -> str:
        """转换字典为HOCON格式"""
        lines = []
        prefix = " " * indent
        for key, value in obj.items():
            if value is None:
                continue
            if isinstance(value, dict):
                lines.append(f"{prefix}{key} {{")
                lines.append(self._dict_to_hocon(value, indent + 4))
                lines.append(f"{prefix}}}")
            elif isinstance(value, list):
                if all(isinstance(item, dict) for item in value):
                    for item in value:
                        lines.append(f"{prefix}{key} = [")
                        lines.append(self._dict_to_hocon(item, indent + 4))
                        lines.append(f"{prefix}]")
                else:
                    items = ', '.join([f'"{i}"' if isinstance(i, str) else str(i) for i in value])
                    lines.append(f"{prefix}{key} = [{items}]")
            else:
                if isinstance(value, bool):
                    lines.append(f"{prefix}{key} = {'true' if value else 'false'}")
                elif isinstance(value, (int, float)):
                    lines.append(f"{prefix}{key} = {value}")
                else:
                    lines.append(f"{prefix}{key} = \"{value}\"")
        return "\n".join(lines)
    def save_config(self, filename: str = "emqx.conf"):
        """保存配置"""
        config_path = self.config_dir / filename
        config_path.parent.mkdir(parents=True, exist_ok=True)
        config_str = self.create_default_config()
        with open(config_path, 'w') as f:
            f.write(config_str)
        print(f"配置已保存到: {config_path}")
        return config_path
    def backup_config(self, config_file: str = "emqx.conf"):
        """备份配置"""
        source_path = self.config_dir / config_file
        if not source_path.exists():
            print(f"配置文件不存在: {source_path}")
            return False
        self.backup_dir.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_path = self.backup_dir / f"{config_file}.{timestamp}.backup"
        with open(source_path, 'r') as source:
            with open(backup_path, 'w') as backup:
                backup.write(source.read())
        print(f"备份已保存到: {backup_path}")
        return True
    def export_config(self, output_file: str = None):
        """导出配置为JSON"""
        if output_file is None:
            output_file = f"emqx_config_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        # 这里应该从实际的EMQX配置中读取
        config_dict = {
            'exported_at': datetime.now().isoformat(),
            'version': '5.0',
            'config': self._read_config()
        }
        with open(output_file, 'w') as f:
            json.dump(config_dict, f, indent=2)
        print(f"配置已导出到: {output_file}")
    def _read_config(self) -> dict:
        """读取当前配置（示例）"""
        config_path = self.config_dir / "emqx.conf"
        if not config_path.exists():
            return {}
        # 简化的配置读取，实际中应该解析HOCON
        with open(config_path, 'r') as f:
            content = f.read()
        return {'raw_content': content}
    def validate_config(self) -> bool:
        """验证配置"""
        config_file = self.config_dir / "emqx.conf"
        if not config_file.exists():
            print("配置文件不存在")
            return False
        validator = EMQXConfigValidator()
        is_valid, errors = validator.validate_config_file(str(config_file))
        if is_valid:
            print("配置验证通过")
            return True
        else:
            print("配置验证失败:")
            for error in errors:
                print(f"  - {error}")
            return False
# 命令行接口
def main():
    parser = argparse.ArgumentParser(description='EMQX配置管理工具')
    parser.add_argument('action', choices=['create', 'backup', 'validate', 'export'],
                       help='操作类型')
    parser.add_argument('--config-dir', default='/etc/emqx',
                       help='EMQX配置目录')
    parser.add_argument('--output', help='输出文件路径')
    args = parser.parse_args()
    manager = EMQXConfigManager(args.config_dir)
    if args.action == 'create':
        manager.save_config(args.output or 'emqx.conf')
    elif args.action == 'backup':
        manager.backup_config()
    elif args.action == 'validate':
        manager.validate_config()
    elif args.action == 'export':
        manager.export_config(args.output)
if __name__ == "__main__":
    main()

使用说明

安装依赖：

pip install pyyaml  # 如果需要YAML支持

基本使用：

# 生成简单配置
config = EMQXConfigGenerator()
config.set_node_config("emqx@127.0.0.1")
config.add_tcp_listener(1883)
config.save_to_file("emqx.conf")

生成集群配置：

cluster = EMQXClusterConfigGenerator()
cluster.add_node("emqx1", "192.168.1.10")
cluster.add_node("emqx2", "192.168.1.11")
cluster.build_cluster_config()

这些工具提供了灵活的配置生成方式,可以根据实际需求进行定制和扩展。

本文地址： https://www.bmjup.com/post/16932.html