Python脚本如何生成VerneMQ配置

wen 实用脚本 2026-07-05 184

本文目录导读：

使用Jinja2模板生成配置
使用Python字典直接生成
多环境配置生成

我来介绍几种Python生成VerneMQ配置的方法：

使用Jinja2模板生成配置

安装依赖

pip install jinja2

创建模板文件 `vernemq_config.j2`

{% raw %}
# VerneMQ Configuration
# Generated by Python at {{ generated_time }}
## Listeners
listener.tcp.default = {{ listener_tcp_default }}
listener.ws.default = {{ listener_ws_default }}
listener.http.metrics = {{ listener_http_metrics }}
## Logging
log.console = {{ log_console }}
log.console.level = {{ log_level }}
log.dir = {{ log_dir }}
## MQTT Settings
allow_anonymous = {{ 'on' if allow_anonymous else 'off' }}
max_client_id_size = {{ max_client_id_size }}
max_message_size = {{ max_message_size }}
## Session settings
persistent_client_expiration = {{ persistent_client_expiration }}
## Retry interval
retry_interval = {{ retry_interval }}
{% if plugins %}
## Plugin settings
{% for plugin in plugins %}
vmq_plugin.{{ plugin.name }} = {{ plugin.path }}
{% endfor %}
{% endif %}
## ACL
{% if acl %}
acl_file = {{ acl_file }}
{% endif %}
## Metrics
vmq_metrics.prometheus = {{ metrics_prometheus }}
{% endraw %}

Python脚本生成配置

from jinja2 import Environment, FileSystemLoader
from datetime import datetime
import os
class VerneMQConfigGenerator:
    def __init__(self, template_dir='.'):
        self.env = Environment(
            loader=FileSystemLoader(template_dir),
            trim_blocks=True,
            lstrip_blocks=True
        )
    def generate_basic_config(self):
        """生成基本配置"""
        config = {
            'generated_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'listener_tcp_default': '0.0.0.0:1883',
            'listener_ws_default': '0.0.0.0:8080',
            'listener_http_metrics': '0.0.0.0:8888',
            'log_console': 'on',
            'log_level': 'info',
            'log_dir': '/var/log/vernemq',
            'allow_anonymous': False,
            'max_client_id_size': 23,
            'max_message_size': 0,  # 0 = unlimited
            'persistent_client_expiration': '1h',
            'retry_interval': 20,
            'plugins': [
                {'name': 'demos', 'path': '/usr/lib/vernemq/plugins/vmq_demos'}
            ],
            'acl': True,
            'acl_file': '/etc/vernemq/acl.conf',
            'metrics_prometheus': 'on'
        }
        return config
    def generate_production_config(self):
        """生成生产环境配置"""
        config = {
            'generated_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'listener_tcp_default': '0.0.0.0:1883',
            'listener_ws_default': '0.0.0.0:8080',
            'listener_http_metrics': '127.0.0.1:8888',  # 仅本地访问
            'log_console': 'on',
            'log_level': 'warning',
            'log_dir': '/var/log/vernemq',
            'allow_anonymous': False,
            'max_client_id_size': 100,
            'max_message_size': 10485760,  # 10MB
            'persistent_client_expiration': '24h',
            'retry_interval': 10,
            'plugins': [
                {'name': 'auth', 'path': '/usr/lib/vernemq/plugins/vmq_auth'},
                {'name': 'bridge', 'path': '/usr/lib/vernemq/plugins/vmq_bridge'},
                {'name': 'metrics', 'path': '/usr/lib/vernemq/plugins/vmq_metrics'}
            ],
            'acl': True,
            'acl_file': '/etc/vernemq/acl.conf',
            'metrics_prometheus': 'on'
        }
        return config
    def generate_cluster_config(self):
        """生成集群配置"""
        base_config = self.generate_production_config()
        cluster_config = {
            **base_config,
            'generated_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            # 集群配置
            'cluster_discovery': 'etcd',
            'cluster_etcd_discovery_ttl': '30',
            'cluster_etcd_key': 'vernemq/cluster',
            'etcd_host': ['http://etcd1:2379', 'http://etcd2:2379', 'http://etcd3:2379'],
            # 更多配置...
        }
        return cluster_config
    def render_config(self, config_data, node_name='node1'):
        """渲染配置"""
        template = self.env.get_template('vernemq_config.j2')
        config_content = template.render(**config_data)
        # 添加节点特定配置
        additional_config = f"""
## Node specific configuration
nodename = {node_name}
distributed_cookie = vmq_secret_cookie
"""
        return config_content + additional_config
    def save_config(self, content, output_path):
        """保存配置文件"""
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        with open(output_path, 'w') as f:
            f.write(content)
        print(f"Configuration saved to {output_path}")
# 使用示例
if __name__ == '__main__':
    generator = VerneMQConfigGenerator()
    # 生成生产配置
    config_data = generator.generate_production_config()
    config_content = generator.render_config(config_data, 'vmq-prod')
    generator.save_config(config_content, '/etc/vernemq/vernemq.conf')
    # 生成集群配置
    cluster_data = generator.generate_cluster_config()
    cluster_content = generator.render_config(cluster_data, 'vmq-cluster-1')
    generator.save_config(cluster_content, '/etc/vernemq/vernemq-cluster.conf')

使用Python字典直接生成

class SimpleVerneMQConfigGenerator:
    def __init__(self):
        self.config = {}
    def set_property(self, key, value):
        """设置配置属性"""
        self.config[key] = value
    def generate_basic(self):
        """基础配置"""
        self.set_property('listener.tcp.default', '0.0.0.0:1883')
        self.set_property('listener.ws.default', '0.0.0.0:8080')
        self.set_property('allow_anonymous', 'off')
        self.set_property('log.console', 'on')
        self.set_property('log.console.level', 'info')
    def generate_advanced(self):
        """高级配置"""
        self.generate_basic()
        self.set_property('max_client_id_size', '100')
        self.set_property('max_message_size', '10485760')
        self.set_property('persistent_client_expiration', '24h')
        self.set_property('retry_interval', '10')
        # ACL配置
        self.set_property('acl_file', '/etc/vernemq/acl.conf')
        # 插件配置
        plugins = ['vmq_demos', 'vmq_bridge', 'vmq_metrics']
        for plugin in plugins:
            self.set_property(f'vmq_plugin.{plugin}', f'/usr/lib/vernemq/plugins/{plugin}')
    def generate_cluster(self):
        """集群配置"""
        self.generate_advanced()
        self.set_property('cluster.discovery', 'etcd')
        self.set_property('cluster.etcd.discovery_ttl', '30')
        self.set_property('listener.http.metrics', '127.0.0.1:8888')
    def to_string(self):
        """转换为配置字符串"""
        config_lines = []
        config_lines.append("# VerneMQ Configuration")
        config_lines.append("# Generated by Python\n")
        for key, value in self.config.items():
            config_lines.append(f"{key} = {value}")
        return '\n'.join(config_lines)
    def save(self, filename):
        """保存配置到文件"""
        with open(filename, 'w') as f:
            f.write(self.to_string())
# 使用示例
config = SimpleVerneMQConfigGenerator()
config.generate_cluster()
config.save('vernemq.conf')

多环境配置生成

import json
import yaml
from typing import Dict, List
class MultiEnvironmentConfigGenerator:
    def __init__(self, base_config_path: str = None):
        self.environments = {
            'dev': self._dev_config(),
            'staging': self._staging_config(),
            'production': self._production_config()
        }
        if base_config_path:
            with open(base_config_path) as f:
                base_config = json.load(f)
                self.environments = {env: {**base_config, **cfg} 
                                   for env, cfg in self.environments.items()}
    def _dev_config(self) -> Dict:
        return {
            'listener.tcp.default': '0.0.0.0:1883',
            'listener.ws.default': '0.0.0.0:8080',
            'log.console': 'on',
            'log.console.level': 'debug',
            'allow_anonymous': 'on',
            'max_client_id_size': '255'
        }
    def _staging_config(self) -> Dict:
        return {
            'listener.tcp.default': '0.0.0.0:1883',
            'listener.ws.default': '0.0.0.0:8080',
            'log.console': 'on',
            'log.console.level': 'info',
            'allow_anonymous': 'off',
            'max_client_id_size': '100',
            'acl_file': '/etc/vernemq/staging.acl'
        }
    def _production_config(self) -> Dict:
        return {
            'listener.tcp.default': '0.0.0.0:1883',
            'listener.ws.default': '0.0.0.0:8080',
            'listener.http.metrics': '127.0.0.1:8888',
            'log.console': 'on',
            'log.console.level': 'warning',
            'allow_anonymous': 'off',
            'max_client_id_size': '100',
            'max_message_size': '10485760',
            'persistent_client_expiration': '24h',
            'retry_interval': '10',
            'acl_file': '/etc/vernemq/acl.conf',
            'cluster.discovery': 'etcd'
        }
    def generate_config(self, environment: str, node_name: str = None) -> str:
        """生成指定环境的配置"""
        if environment not in self.environments:
            raise ValueError(f"Unknown environment: {environment}")
        config = self.environments[environment].copy()
        config['generated_by'] = 'Python MultiEnvironmentConfigGenerator'
        config['environment'] = environment
        if node_name:
            config['nodename'] = node_name
        # 转换为配置字符串
        config_lines = []
        config_lines.append(f"# VerneMQ Configuration - {environment} environment")
        config_lines.append(f"# Generated: {datetime.now().isoformat()}\n")
        for key, value in config.items():
            if isinstance(value, list):
                for item in value:
                    config_lines.append(f"{key} = {item}")
            else:
                config_lines.append(f"{key} = {value}")
        return '\n'.join(config_lines)
    def save_all(self, output_dir: str):
        """保存所有环境的配置"""
        for env, config in self.environments.items():
            config_content = self.generate_config(env)
            output_path = os.path.join(output_dir, f'vernemq.{env}.conf')
            with open(output_path, 'w') as f:
                f.write(config_content)
            print(f"Generated: {output_path}")
# 使用示例
generator = MultiEnvironmentConfigGenerator()
generator.save_all('./configs')

这些方法提供了不同层次的灵活性和功能：