实用脚本能自动发送RabbitMQ消息吗？

wen 实用脚本 2026-07-21 169

本文目录导读：

最简单的Python脚本（使用pika库）
Shell脚本（使用rabbitmqadmin）
Node.js脚本
配置文件（config.json）
使用方法
自动化场景示例

是的,完全可以！下面我给你几个实用脚本，能自动发送RabbitMQ消息。

最简单的Python脚本（使用pika库）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""自动发送RabbitMQ消息脚本"""
import pika
import json
import time
import random
from datetime import datetime
class RabbitMQSender:
    def __init__(self, host='localhost', port=5672, 
                 username='guest', password='guest',
                 virtual_host='/'):
        """初始化RabbitMQ连接"""
        credentials = pika.PlainCredentials(username, password)
        parameters = pika.ConnectionParameters(
            host=host,
            port=port,
            virtual_host=virtual_host,
            credentials=credentials
        )
        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()
    def send_to_queue(self, queue_name, message, durable=True):
        """发送消息到指定队列"""
        # 声明队列（如果不存在则创建）
        self.channel.queue_declare(queue=queue_name, durable=durable)
        # 发送消息
        self.channel.basic_publish(
            exchange='',
            routing_key=queue_name,
            body=json.dumps(message),
            properties=pika.BasicProperties(
                delivery_mode=2,  # 持久化消息
                content_type='application/json',
                timestamp=int(time.time())
            )
        )
        print(f"[✓] 消息已发送到队列 [{queue_name}]: {message}")
    def send_to_exchange(self, exchange_name, routing_key, message, 
                         exchange_type='direct'):
        """发送消息到交换机"""
        # 声明交换机
        self.channel.exchange_declare(
            exchange=exchange_name, 
            exchange_type=exchange_type,
            durable=True
        )
        # 发送消息
        self.channel.basic_publish(
            exchange=exchange_name,
            routing_key=routing_key,
            body=json.dumps(message),
            properties=pika.BasicProperties(
                delivery_mode=2,
                content_type='application/json'
            )
        )
        print(f"[✓] 消息已发送到交换机 [{exchange_name}] 路由键 [{routing_key}]")
    def close(self):
        """关闭连接"""
        if self.connection and not self.connection.is_closed:
            self.connection.close()
            print("[✓] 连接已关闭")
def main():
    """示例：自动发送不同类型的消息"""
    sender = RabbitMQSender()
    try:
        # 示例1：发送普通消息到队列
        queue_message = {
            'id': random.randint(1000, 9999),
            'type': 'user_action',
            'data': {
                'user_id': 'U001',
                'action': 'login',
                'timestamp': datetime.now().isoformat()
            }
        }
        sender.send_to_queue('user_actions', queue_message)
        # 示例2：发送到交换机（fanout模式）
        broadcast_message = {
            'event': 'system_notification',
            'message': '系统维护通知',
            'level': 'info',
            'timestamp': datetime.now().isoformat()
        }
        sender.send_to_exchange(
            'system_notifications', 
            '',  # fanout交换机不需要routing_key
            broadcast_message,
            exchange_type='fanout'
        )
        # 示例3：发送到topic交换机
        topic_message = {
            'order_id': f'ORD{random.randint(10000, 99999)}',
            'status': 'completed',
            'amount': round(random.uniform(10, 1000), 2)
        }
        sender.send_to_exchange(
            'order_events', 
            'order.completed',  # routing key
            topic_message,
            exchange_type='topic'
        )
        # 示例4：批量发送消息（模拟实时数据）
        print("\n--- 开始批量发送消息 ---")
        for i in range(5):
            sensor_data = {
                'sensor_id': f'SENSOR-{random.randint(1, 10):03d}',
                'temperature': round(random.uniform(20, 30), 1),
                'humidity': round(random.uniform(40, 60), 1),
                'reading_time': datetime.now().isoformat()
            }
            sender.send_to_queue('sensor_data', sensor_data)
            time.sleep(0.5)  # 间隔0.5秒发送
    except Exception as e:
        print(f"[✗] 发送失败: {e}")
    finally:
        sender.close()
if __name__ == '__main__':
    main()

Shell脚本（使用rabbitmqadmin）

#!/bin/bash
# 自动发送RabbitMQ消息的Shell脚本
# 配置
RABBIT_HOST="localhost"
RABBIT_PORT="15672"
RABBIT_USER="guest"
RABBIT_PASS="guest"
QUEUE_NAME="test_queue"
EXCHANGE_NAME="test_exchange"
# 发送消息到队列
send_to_queue() {
    local message="$1"
    local queue="${2:-$QUEUE_NAME}"
    rabbitmqadmin \
        -H "$RABBIT_HOST" \
        -P "$RABBIT_PORT" \
        -u "$RABBIT_USER" \
        -p "$RABBIT_PASS" \
        publish \
        exchange=amq.default \
        routing_key="$queue" \
        payload="$message"
    echo "[✓] 消息已发送到队列 [$queue]: $message"
}
# 发送JSON消息到交换机
send_json_to_exchange() {
    local message="$1"
    local routing_key="$2"
    local exchange="${3:-$EXCHANGE_NAME}"
    local exchange_type="${4:-direct}"
    # 声明交换机
    rabbitmqadmin \
        -H "$RABBIT_HOST" \
        -P "$RABBIT_PORT" \
        -u "$RABBIT_USER" \
        -p "$RABBIT_PASS" \
        declare exchange \
        name="$exchange" \
        type="$exchange_type" \
        durable=true
    # 发送消息
    rabbitmqadmin \
        -H "$RABBIT_HOST" \
        -P "$RABBIT_PORT" \
        -u "$RABBIT_USER" \
        -p "$RABBIT_PASS" \
        publish \
        exchange="$exchange" \
        routing_key="$routing_key" \
        payload="$message" \
        properties='{"content_type":"application/json"}'
    echo "[✓] JSON消息已发送到 [$exchange] 路由键 [$routing_key]"
}
# 批量发送消息
batch_send() {
    local count="${1:-10}"
    local queue="${2:-$QUEUE_NAME}"
    echo "--- 开始批量发送 $count 条消息到 [$queue] ---"
    for ((i=1; i<=count; i++)); do
        local message="{\"id\":$i,\"timestamp\":\"$(date -u +"%Y-%m-%dT%H:%M:%SZ")\",\"data\":\"消息$i\"}"
        send_to_queue "$message" "$queue"
        sleep 0.2
    done
    echo "--- 批量发送完成 ---"
}
# 主函数
main() {
    # 测试发送
    send_to_queue '{"type":"test","message":"Hello RabbitMQ!"}'
    # 发送JSON到交换机
    send_json_to_exchange \
        '{"event":"user_login","user_id":"U001"}' \
        'user.login' \
        'user_events' \
        'topic'
    # 批量发送
    batch_send 5
}
# 执行主函数
main

Node.js脚本

#!/usr/bin/env node
// 自动发送RabbitMQ消息脚本
const amqp = require('amqplib');
const crypto = require('crypto');
class RabbitMQSender {
    constructor(config = {}) {
        this.config = {
            host: config.host || 'localhost',
            port: config.port || 5672,
            user: config.user || 'guest',
            password: config.password || 'guest',
            vhost: config.vhost || '/'
        };
        this.connection = null;
        this.channel = null;
    }
    async connect() {
        const url = `amqp://${this.config.user}:${this.config.password}@${this.config.host}:${this.config.port}${this.config.vhost}`;
        this.connection = await amqp.connect(url);
        this.channel = await this.connection.createChannel();
        console.log('[✓] 已连接到RabbitMQ');
    }
    async sendToQueue(queueName, message, durable = true) {
        await this.channel.assertQueue(queueName, { durable });
        this.channel.sendToQueue(
            queueName,
            Buffer.from(JSON.stringify(message)),
            {
                persistent: true,
                contentType: 'application/json',
                timestamp: Date.now()
            }
        );
        console.log(`[✓] 消息已发送到队列 [${queueName}]:`, message);
    }
    async sendToExchange(exchangeName, routingKey, message, exchangeType = 'direct') {
        await this.channel.assertExchange(exchangeName, exchangeType, { durable: true });
        this.channel.publish(
            exchangeName,
            routingKey,
            Buffer.from(JSON.stringify(message)),
            {
                persistent: true,
                contentType: 'application/json'
            }
        );
        console.log(`[✓] 消息已发送到交换机 [${exchangeName}] 路由键 [${routingKey}]`);
    }
    async close() {
        if (this.connection) {
            await this.connection.close();
            console.log('[✓] 连接已关闭');
        }
    }
}
// 使用示例
async function main() {
    const sender = new RabbitMQSender();
    try {
        await sender.connect();
        // 发送订单消息
        await sender.sendToQueue('orders', {
            orderId: `ORD-${crypto.randomUUID().slice(0, 8)}`,
            userId: 'U001',
            amount: 99.99,
            status: 'pending',
            timestamp: new Date().toISOString()
        });
        // 发送到topic交换机
        await sender.sendToExchange(
            'events',
            'order.created',
            {
                event: 'order_created',
                data: { orderId: 'ORD-12345', amount: 199.99 }
            },
            'topic'
        );
        // 批量发送
        console.log('\n--- 开始批量发送 ---');
        for (let i = 0; i < 3; i++) {
            await sender.sendToQueue('logs', {
                level: 'info',
                message: `日志消息 #${i + 1}`,
                timestamp: new Date().toISOString()
            });
        }
    } catch (error) {
        console.error('[✗] 发送失败:', error.message);
    } finally {
        await sender.close();
    }
}
main();

配置文件（config.json）

{
    "rabbitmq": {
        "host": "localhost",
        "port": 5672,
        "username": "guest",
        "password": "guest",
        "virtual_host": "/"
    },
    "queues": {
        "orders": {
            "durable": true,
            "exclusive": false,
            "auto_delete": false
        },
        "logs": {
            "durable": true
        }
    },
    "exchanges": {
        "events": {
            "type": "topic",
            "durable": true
        },
        "notifications": {
            "type": "fanout",
            "durable": true
        }
    }
}

使用方法

Python版本：

# 安装依赖
pip install pika
# 运行脚本
python rabbitmq_sender.py

Shell版本：

# 设置执行权限
chmod +x rabbitmq_sender.sh
# 运行脚本
./rabbitmq_sender.sh

Node.js版本：

# 安装依赖
npm install amqplib
# 运行脚本
node rabbitmq_sender.js

自动化场景示例

你可以将这些脚本用于：

定时任务：配合cron定期发送消息
数据同步：自动发送数据库变更消息
监控告警：发送系统监控数据
批量导入：批量导入数据到消息队列
测试环境：自动生成测试数据

这些脚本都非常实用,可以满足大部分自动发送RabbitMQ消息的需求，你需要哪个版本的？我可以帮你修改得更适合你的具体场景。

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