Python脚本如何生成RabbitMQ配置

wen 实用脚本 2026-07-06 186

本文目录导读：

使用Pika库生成配置
使用PyRabbit库生成配置文件
生成可执行的配置脚本
配置模板生成器
配置文件生成器工具类

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

使用Pika库生成配置

import pika
import json
class RabbitMQConfigGenerator:
    def __init__(self, host='localhost', port=5672, vhost='/'):
        self.connection_params = pika.ConnectionParameters(
            host=host,
            port=port,
            virtual_host=vhost,
            credentials=pika.PlainCredentials('guest', 'guest')
        )
    def generate_exchange_config(self, exchange_name, exchange_type='direct', 
                                  durable=True, auto_delete=False):
        """生成交换机配置"""
        return {
            'exchange': exchange_name,
            'exchange_type': exchange_type,
            'durable': durable,
            'auto_delete': auto_delete,
            'arguments': {}
        }
    def generate_queue_config(self, queue_name, durable=True, 
                              exclusive=False, auto_delete=False):
        """生成队列配置"""
        return {
            'queue': queue_name,
            'durable': durable,
            'exclusive': exclusive,
            'auto_delete': auto_delete,
            'arguments': {
                'x-message-ttl': 60000,  # 消息存活时间(毫秒)
                'x-max-length': 10000     # 最大队列长度
            }
        }
    def generate_binding_config(self, queue_name, exchange_name, 
                                 routing_key=''):
        """生成绑定配置"""
        return {
            'queue': queue_name,
            'exchange': exchange_name,
            'routing_key': routing_key
        }
    def save_config_to_json(self, config, filename='rabbitmq_config.json'):
        """保存配置到JSON文件"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
# 使用示例
generator = RabbitMQConfigGenerator()
# 生成完整配置
full_config = {
    'connection': {
        'host': 'localhost',
        'port': 5672,
        'vhost': '/',
        'credentials': {
            'username': 'guest',
            'password': 'guest'
        }
    },
    'exchanges': [
        generator.generate_exchange_config('order.exchange', 'topic'),
        generator.generate_exchange_config('notification.exchange', 'fanout')
    ],
    'queues': [
        generator.generate_queue_config('order.queue'),
        generator.generate_queue_config('notification.queue',
                                         arguments={
                                             'x-message-ttl': 30000,
                                             'x-dead-letter-exchange': 'dead.letter.exchange'
                                         })
    ],
    'bindings': [
        generator.generate_binding_config('order.queue', 'order.exchange', 'order.#'),
        generator.generate_binding_config('notification.queue', 'notification.exchange')
    ]
}
generator.save_config_to_json(full_config)

使用PyRabbit库生成配置文件

import pyrabbit
import yaml
class RabbitMQConfigFromServer:
    def __init__(self, host='localhost', port=15672, 
                 user='guest', password='guest'):
        self.client = pyrabbit.Client(f'http://{host}:{port}', user, password)
    def generate_config_from_server(self):
        """从现有RabbitMQ服务器生成配置"""
        config = {
            'rabbitmq': {
                'vhosts': [],
                'exchanges': [],
                'queues': [],
                'bindings': []
            }
        }
        # 获取虚拟主机
        vhosts = self.client.get_vhosts()
        for vhost in vhosts:
            vhost_config = {
                'name': vhost['name'],
                'tracing': vhost['tracing']
            }
            config['rabbitmq']['vhosts'].append(vhost_config)
            # 获取该vhost下的交换机
            exchanges = self.client.get_exchanges(vhost['name'])
            for exchange in exchanges:
                if not exchange['name'].startswith('amq.'):  # 排除内置交换机
                    exchange_config = {
                        'name': exchange['name'],
                        'type': exchange['type'],
                        'durable': exchange['durable'],
                        'auto_delete': exchange['auto_delete'],
                        'arguments': exchange.get('arguments', {})
                    }
                    config['rabbitmq']['exchanges'].append(exchange_config)
        return config
    def save_config_to_yaml(self, config, filename='rabbitmq_config.yaml'):
        """保存为YAML格式"""
        with open(filename, 'w', encoding='utf-8') as f:
            yaml.dump(config, f, default_flow_style=False, allow_unicode=True)
# 使用示例
# config_generator = RabbitMQConfigFromServer()
# config = config_generator.generate_config_from_server()
# config_generator.save_config_to_yaml(config)

生成可执行的配置脚本

class RabbitMQSetupScriptGenerator:
    @staticmethod
    def generate_python_setup_script(config):
        """生成Python配置脚本"""
        script = """
import pika
def setup_rabbitmq():
    # 连接RabbitMQ
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(
            host='{host}',
            port={port},
            virtual_host='{vhost}',
            credentials=pika.PlainCredentials('{username}', '{password}')
        )
    )
    channel = connection.channel()
    # 创建交换机
""".format(**config['connection'])
        # 添加交换机创建代码
        for exchange in config.get('exchanges', []):
            script += f"""
    channel.exchange_declare(
        exchange='{exchange['exchange']}',
        exchange_type='{exchange.get('exchange_type', 'direct')}',
        durable={exchange.get('durable', True)},
        auto_delete={exchange.get('auto_delete', False)}
    )
"""
        # 添加队列创建代码
        for queue in config.get('queues', []):
            arguments = queue.get('arguments', {})
            script += f"""
    channel.queue_declare(
        queue='{queue['queue']}',
        durable={queue.get('durable', True)},
        exclusive={queue.get('exclusive', False)},
        auto_delete={queue.get('auto_delete', False)},
        arguments={arguments}
    )
"""
        # 添加绑定代码
        for binding in config.get('bindings', []):
            script += f"""
    channel.queue_bind(
        queue='{binding['queue']}',
        exchange='{binding['exchange']}',
        routing_key='{binding.get('routing_key', '')}'
    )
"""
        script += """
    connection.close()
    print("RabbitMQ配置已成功应用！")
if __name__ == '__main__':
    setup_rabbitmq()
"""
        return script
    @staticmethod
    def generate_bash_setup_script(config):
        """生成Bash配置脚本(使用rabbitmqctl)"""
        script = "#!/bin/bash\n\n"
        # 添加vhost配置
        for vhost in config.get('vhosts', []):
            script += f"rabbitmqctl add_vhost {vhost.get('name', '/')}\n"
        # 添加用户配置
        for user in config.get('users', []):
            script += f"rabbitmqctl add_user {user['username']} {user['password']}\n"
            script += f"rabbitmqctl set_permissions -p / {user['username']} \".*\" \".*\" \".*\"\n"
        # 添加交换机配置
        for exchange in config.get('exchanges', []):
            script += f"""
rabbitmqadmin declare exchange \\
    name={exchange['exchange']} \\
    type={exchange.get('exchange_type', 'direct')} \\
    durable={str(exchange.get('durable', True)).lower()}
"""
        # 添加队列配置
        for queue in config.get('queues', []):
            script += f"""
rabbitmqadmin declare queue \\
    name={queue['queue']} \\
    durable={str(queue.get('durable', True)).lower()}
"""
        # 添加绑定配置
        for binding in config.get('bindings', []):
            script += f"""
rabbitmqadmin declare binding \\
    source={binding['exchange']} \\
    destination={binding['queue']} \\
    routing_key={binding.get('routing_key', '')}
"""
        return script
# 使用示例
generator = RabbitMQSetupScriptGenerator()
config = {
    'connection': {
        'host': 'localhost', 'port': 5672, 'vhost': '/',
        'username': 'admin', 'password': 'admin123'
    },
    'exchanges': [
        {'exchange': 'order.exchange', 'exchange_type': 'topic'},
        {'exchange': 'log.exchange', 'exchange_type': 'fanout'}
    ],
    'queues': [
        {'queue': 'order.queue', 'durable': True},
        {'queue': 'log.queue', 'arguments': {'x-message-ttl': 3600000}}
    ],
    'bindings': [
        {'queue': 'order.queue', 'exchange': 'order.exchange', 'routing_key': 'order.*'},
        {'queue': 'log.queue', 'exchange': 'log.exchange'}
    ]
}
# 生成Python配置脚本
python_script = generator.generate_python_setup_script(config)
with open('setup_rabbitmq.py', 'w') as f:
    f.write(python_script)
# 生成Bash配置脚本
bash_script = generator.generate_bash_setup_script(config)
with open('setup_rabbitmq.sh', 'w') as f:
    f.write(bash_script)

配置模板生成器

from string import Template
import os
class RabbitMQTemplateConfig:
    def __init__(self, template_dir='templates'):
        self.template_dir = template_dir
        os.makedirs(template_dir, exist_ok=True)
    def create_pika_template(self):
        """创建Pika配置模板"""
        template = Template("""
import pika
class RabbitMQConfig:
    def __init__(self):
        self.connection_params = {
            'host': '${host}',
            'port': ${port},
            'virtual_host': '${vhost}',
            'credentials': pika.PlainCredentials('${username}', '${password}')
        }
        self.exchanges = ${exchanges}
        self.queues = ${queues}
        self.bindings = ${bindings}
    def apply_config(self):
        connection = pika.BlockingConnection(
            pika.ConnectionParameters(**self.connection_params)
        )
        channel = connection.channel()
        # 应用配置
        for exchange in self.exchanges:
            channel.exchange_declare(**exchange)
        for queue in self.queues:
            channel.queue_declare(**queue)
        for binding in self.bindings:
            channel.queue_bind(**binding)
        connection.close()
    def validate_config(self):
        # 配置验证逻辑
        return True
""")
        template_path = os.path.join(self.template_dir, 'pika_config_template.py')
        with open(template_path, 'w') as f:
            f.write(template.template)
        return template_path
    def generate_config_from_template(self, params):
        """根据模板生成配置文件"""
        # 读取模板
        template_path = os.path.join(self.template_dir, 'pika_config_template.py')
        with open(template_path, 'r') as f:
            template_content = f.read()
        template = Template(template_content)
        # 填充参数
        config_content = template.safe_substitute(params)
        # 保存生成的配置
        output_path = f'config_{params.get("host", "localhost")}.py'
        with open(output_path, 'w') as f:
            f.write(config_content)
        return output_path
# 使用示例
template_gen = RabbitMQTemplateConfig()
params = {
    'host': 'rabbitmq.production.com',
    'port': 5672,
    'vhost': '/production',
    'username': 'prod_user',
    'password': 'secure_password',
    'exchanges': "[{'exchange': 'main.exchange', 'exchange_type': 'topic'}]",
    'queues': "[{'queue': 'main.queue', 'durable': True}]",
    'bindings': "[{'queue': 'main.queue', 'exchange': 'main.exchange', 'routing_key': '#'}]"
}
# 生成配置文件
config_file = template_gen.generate_config_from_template(params)

配置文件生成器工具类

import json
import yaml
import configparser
from typing import Dict, Any
class RabbitMQConfigFileGenerator:
    def __init__(self):
        self.config = {
            'connection': {},
            'vhosts': [],
            'users': [],
            'permissions': [],
            'policies': [],
            'exchanges': [],
            'queues': [],
            'bindings': []
        }
    def add_connection(self, host='localhost', port=5672, 
                       vhost='/', username='guest', password='guest'):
        self.config['connection'] = {
            'host': host,
            'port': port,
            'vhost': vhost,
            'credentials': {
                'username': username,
                'password': password
            }
        }
    def add_exchange(self, name, exchange_type='direct', 
                     durable=True, auto_delete=False, arguments=None):
        self.config['exchanges'].append({
            'name': name,
            'type': exchange_type,
            'durable': durable,
            'auto_delete': auto_delete,
            'arguments': arguments or {}
        })
    def add_queue(self, name, durable=True, exclusive=False, 
                  auto_delete=False, arguments=None):
        self.config['queues'].append({
            'name': name,
            'durable': durable,
            'exclusive': exclusive,
            'auto_delete': auto_delete,
            'arguments': arguments or {}
        })
    def add_binding(self, queue, exchange, routing_key='', 
                    arguments=None):
        self.config['bindings'].append({
            'queue': queue,
            'exchange': exchange,
            'routing_key': routing_key,
            'arguments': arguments or {}
        })
    def save_as_json(self, filename='rabbitmq_config.json'):
        """保存为JSON格式"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.config, f, indent=2, ensure_ascii=False)
        print(f"配置已保存到 {filename}")
    def save_as_yaml(self, filename='rabbitmq_config.yaml'):
        """保存为YAML格式"""
        with open(filename, 'w', encoding='utf-8') as f:
            yaml.dump(self.config, f, default_flow_style=False, 
                     allow_unicode=True)
        print(f"配置已保存到 {filename}")
    def save_as_ini(self, filename='rabbitmq_config.ini'):
        """保存为INI格式"""
        config = configparser.ConfigParser()
        # 连接配置
        config['connection'] = self.config['connection']
        # 其他配置转为字符串列表
        for section in ['exchanges', 'queues', 'bindings']:
            config[section] = {}
            for i, item in enumerate(self.config[section]):
                config[section][f'item_{i}'] = str(item)
        with open(filename, 'w') as f:
            config.write(f)
        print(f"配置已保存到 {filename}")
# 使用示例
gen = RabbitMQConfigFileGenerator()
gen.add_connection('localhost', 5672, '/', 'admin', 'admin123')
gen.add_exchange('order.exchange', 'topic', durable=True)
gen.add_exchange('notification.exchange', 'fanout')
gen.add_queue('order.queue', durable=True, 
              arguments={'x-message-ttl': 60000})
gen.add_queue('notification.queue')
gen.add_binding('order.queue', 'order.exchange', 'order.#')
gen.add_binding('notification.queue', 'notification.exchange')
# 导出为不同格式
gen.save_as_json()
gen.save_as_yaml()
gen.save_as_ini()

这些方法可以根据你的需求生成不同格式的RabbitMQ配置文件,从简单的JSON到可执行的配置脚本都有提供。

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