本文目录导读：
我来为您介绍如何使用Python脚本生成EMQX集群配置。
基础配置生成
1 单节点配置模板
import yaml
import json
import os
from typing import List, Dict, Any
class EMQXConfigGenerator:
def __init__(self, cluster_name: str = "emqx-cluster"):
self.cluster_name = cluster_name
self.config = {
"node": {
"name": "",
"cookie": "",
"data_dir": "data",
"log_dir": "log"
},
"cluster": {
"name": cluster_name,
"discovery_strategy": "manual",
"core_nodes": [],
"replicant_nodes": []
}
}
def set_node_cookie(self, cookie: str):
"""设置Erlang cookie"""
self.config["node"]["cookie"] = cookie
def add_node(self, node_name: str, node_type: str = "core", host: str = "localhost", port: int = 1883):
"""添加节点"""
node_config = {
"name": f"{node_name}@{host}",
"type": node_type,
"listen_port": port,
"listener": {
"tcp": port,
"ssl": port + 1,
"ws": port + 2
}
}
if node_type == "core":
self.config["cluster"]["core_nodes"].append(node_config)
else:
self.config["cluster"]["replicant_nodes"].append(node_config)
return node_config
def generate_yaml(self, output_path: str = "emqx.yaml"):
"""生成YAML配置"""
with open(output_path, 'w') as f:
yaml.dump(self.config, f, default_flow_style=False)
print(f"配置已生成: {output_path}")
def generate_json(self, output_path: str = "emqx.json"):
"""生成JSON配置"""
with open(output_path, 'w') as f:
json.dump(self.config, f, indent=2)
print(f"配置已生成: {output_path}")
自动发现配置生成
1 基于ETCD的自动发现
class ETCDDiscoveryConfig:
def __init__(self):
self.discovery_config = {
"cluster": {
"discovery_strategy": "etcd",
"etcd": {
"server": "http://127.0.0.1:2379",
"prefix": "emqx/",
"node_ttl": "1m",
"ssl": {
"enable": False,
"certfile": "",
"keyfile": ""
}
}
}
}
def set_etcd_servers(self, servers: List[str]):
"""设置ETCD服务器列表"""
self.discovery_config["cluster"]["etcd"]["server"] = ",".join(servers)
def enable_ssl(self, cert_file: str, key_file: str):
"""启用SSL"""
ssl_config = self.discovery_config["cluster"]["etcd"]["ssl"]
ssl_config["enable"] = True
ssl_config["certfile"] = cert_file
ssl_config["keyfile"] = key_file
class DNSDiscoveryConfig:
def __init__(self):
self.discovery_config = {
"cluster": {
"discovery_strategy": "dns",
"dns": {
"record_type": "a",
"name": "emqx-cluster.local",
"app": "emqx"
}
}
}
def set_dns_record(self, record_type: str, name: str, app: str = "emqx"):
"""设置DNS记录"""
self.discovery_config["cluster"]["dns"]["record_type"] = record_type
self.discovery_config["cluster"]["dns"]["name"] = name
self.discovery_config["cluster"]["dns"]["app"] = app
完整集群配置生成
class EMQXClusterConfigGenerator:
def __init__(self):
self.generators = []
self.cluster_config = {
"emqx_cluster": {
"version": "5.0",
"nodes": [],
"discovery": {},
"security": {}
}
}
def add_node_generator(self, node_config: dict):
"""添加节点生成器"""
self.generators.append(node_config)
def generate_multi_node_config(self,
node_count: int = 3,
base_name: str = "emqx",
base_port: int = 1883,
discovery: str = "manual"):
"""生成多节点配置"""
nodes = []
for i in range(node_count):
node = {
"name": f"{base_name}@{base_name}-node-{i+1}",
"type": "core" if i < 2 else "replicant",
"listen_port": base_port + i,
"listener": {
"tcp": base_port + i,
"ssl": base_port + i + 100,
"ws": base_port + i + 200,
"dashboard": 18083 + i
},
"persistent": True if i < 2 else False
}
nodes.append(node)
self.cluster_config["emqx_cluster"]["nodes"] = nodes
self.cluster_config["emqx_cluster"]["discovery"] = {
"strategy": discovery
}
return nodes
def add_security_config(self,
enable_auth: bool = True,
enable_acl: bool = True,
jwt_secret: str = ""):
"""添加安全配置"""
self.cluster_config["emqx_cluster"]["security"] = {
"authentication": {
"enable": enable_auth,
"backend": "built-in-database",
"mechanism": "password-based"
},
"authorization": {
"enable": enable_acl,
"sources": ["built-in-database"]
}
}
if jwt_secret:
self.cluster_config["emqx_cluster"]["security"]["authentication"][
"jwt"] = {
"secret": jwt_secret,
"verify_claims": ["sub", "exp"]
}
def generate_docker_compose(self, output_path: str = "docker-compose.yml"):
"""生成Docker Compose配置"""
compose_config = {
"version": "3.8",
"services": {}
}
for i, node in enumerate(self.cluster_config["emqx_cluster"]["nodes"]):
service_name = f"emqx-node-{i+1}"
compose_config["services"][service_name] = {
"image": "emqx/emqx:5.0",
"container_name": service_name,
"environment": [
f"EMQX_NODE_NAME={node['name']}",
f"EMQX_LISTENER__TCP__DEFAULT={node['listen_port']}",
f"EMQX_CLUSTER__DISCOVERY_STRATEGY=manual",
f"EMQX_CLUSTER__STATIC__SEEDS=emqx@emqx-node-1,emqx@emqx-node-2"
],
"ports": [
f"{node['listen_port']}:{node['listen_port']}",
f"{node['listener']['dashboard']}:18083"
],
"volumes": [
f"./emqx-{i+1}.conf:/opt/emqx/etc/emqx.conf"
]
}
with open(output_path, 'w') as f:
yaml.dump(compose_config, f, default_flow_style=False)
print(f"Docker Compose配置已生成: {output_path}")
def generate_kubernetes_config(self, output_path: str = "emqx-cluster.yaml"):
"""生成Kubernetes配置"""
k8s_config = {
"apiVersion": "apps/v1",
"kind": "StatefulSet",
"metadata": {
"name": "emqx-cluster",
"labels": {
"app": "emqx"
}
},
"spec": {
"serviceName": "emqx-headless",
"replicas": len(self.cluster_config["emqx_cluster"]["nodes"]),
"selector": {
"matchLabels": {
"app": "emqx"
}
},
"template": {
"metadata": {
"labels": {
"app": "emqx"
}
},
"spec": {
"containers": [{
"name": "emqx",
"image": "emqx/emqx:5.0",
"ports": [
{"containerPort": 1883, "name": "mqtt"},
{"containerPort": 8883, "name": "mqtt-ssl"},
{"containerPort": 8083, "name": "ws"},
{"containerPort": 18083, "name": "dashboard"}
],
"env": [
{"name": "EMQX_CLUSTER__DISCOVERY_STRATEGY", "value": "dns"},
{"name": "EMQX_CLUSTER__DNS__RECORD_TYPE", "value": "a"},
{"name": "EMQX_CLUSTER__DNS__NAME", "value": "emqx-headless.default.svc.cluster.local"}
]
}]
}
}
}
}
with open(output_path, 'w') as f:
yaml.dump(k8s_config, f, default_flow_style=False)
print(f"Kubernetes配置已生成: {output_path}")
配置验证工具
class EMQXConfigValidator:
@staticmethod
def validate_cluster_config(config: dict) -> List[str]:
"""验证集群配置"""
errors = []
# 验证节点配置
if "emqx_cluster" not in config:
errors.append("缺少 emqx_cluster 主节点配置")
return errors
nodes = config["emqx_cluster"].get("nodes", [])
if len(nodes) == 0:
errors.append("至少需要一个节点")
# 检查节点名称唯一性
node_names = [node["name"].split("@")[0] for node in nodes]
if len(node_names) != len(set(node_names)):
errors.append("节点名称必须唯一")
# 检查端口冲突
ports = [node["listen_port"] for node in nodes]
if len(ports) != len(set(ports)):
errors.append("端口不能冲突")
# 验证发现策略
discovery = config["emqx_cluster"].get("discovery", {})
if not discovery.get("strategy"):
errors.append("必须指定发现策略")
return errors
@staticmethod
def validate_network_config(config: dict) -> List[str]:
"""验证网络配置"""
errors = []
for node in config.get("emqx_cluster", {}).get("nodes", []):
# 验证端口范围
for port_type in ["tcp", "ssl", "ws", "dashboard"]:
port = node.get("listener", {}).get(port_type, 0)
if port < 1024 or port > 65535:
errors.append(f"节点 {node['name']} 的 {port_type} 端口无效: {port}")
return errors
使用示例
def main():
# 创建配置生成器
generator = EMQXClusterConfigGenerator()
# 生成3节点配置
nodes = generator.generate_multi_node_config(
node_count=3,
base_name="emqx-cluster",
base_port=1883,
discovery="manual"
)
# 添加安全配置
generator.add_security_config(
enable_auth=True,
enable_acl=True,
jwt_secret="your-jwt-secret"
)
# 生成不同格式的配置
generator.generate_docker_compose()
generator.generate_kubernetes_config()
# 验证配置
validator = EMQXConfigValidator()
errors = validator.validate_cluster_config(generator.cluster_config)
if errors:
print("配置验证失败:")
for error in errors:
print(f" - {error}")
else:
print("配置验证成功!")
# 打印生成的节点信息
for node in nodes:
print(f"节点: {node['name']}")
print(f" 类型: {node['type']}")
print(f" 端口: {node['listen_port']}")
print(f" Dashboard: {node['listener']['dashboard']}")
print()
if __name__ == "__main__":
main()
这个Python脚本可以：
- 自动生成EMQX集群配置（YAML/JSON格式）
- 支持多种发现策略（manual、etcd、dns）
- 生成Docker Compose和K8s配置
- 配置验证确保配置正确
- 支持安全配置（认证、授权、JWT）
您可以根据实际需求调整配置参数。
