本文目录导读：
我来详细介绍如何使用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()
这些工具提供了灵活的配置生成方式,可以根据实际需求进行定制和扩展。
