[
    {rabbit, [
        % 强制队列同步
        {force_queue_sync, false},
        
        % 消息存储写入策略
        {msg_store_write_strategy, write_back},
        
        % 同步事务超时
        {sync_tx_timeout, 30000},
        
        % 队列索引最大段数
        {queue_index_max_segment_size, 1024},
        
        % 消息存储文件大小阈值
        {msg_store_file_size_limit, 16777216},
        
        % 日志格式（json）
        {log_formatter, {logger_formatter, #{single_line => true, template => [time, " ", pid, " ", level, ": ", msg, "\n"]}}}
    ]},
    {rabbitmq_management, [
        % Management API 跨域配置
        {cors_allow_origins, ["*"]},
        % Management 接口反向代理
        {reverse_proxy, false}
    ]},
    {rabbitmq_auth_backend_ldap, [
        % LDAP 认证配置
        {servers, ["ldap.example.com"]},
        {user_dn_pattern, "cn=${username},ou=users,dc=example,dc=com"},
        {use_ssl, true},
        {port, 636}
    ]},
    {rabbitmq_shovel, [
        {shovels, [
            {my_shovel, [
                {sources, [{broker, "amqp://source"}]},
                {destinations, [{broker, "amqp://dest"}]},
                {queue, "source_queue"},
                {destination_queue, "dest_queue"},
                {ack_mode, on_confirm}
            ]}
        ]}
    ]}
].

# ===== 连接与网络 =====
listeners.tcp.default = 5672
num_acceptors.tcp = 30
max_connections = 10000
heartbeat = 30
handshake_timeout = 10000
frame_max = 131072
channel_max = 2047

# ===== 内存与磁盘 =====
vm_memory_high_watermark.relative = 0.6
vm_memory_high_watermark_paging_ratio = 0.5
disk_free_limit.absolute = 1GB
disk_free_limit.interval = 5000

# ===== 集群与高可用 =====
cluster_partition_handling = autoheal
cluster_keepalive_interval = 10000
queue_master_locator = min-masters
delegate_count = 32

# ===== 日志 =====
log.file.level = warning
log.file.formatter = json
log.file.rotation.size = 500MB
log.file.rotation.count = 7
reverse_dns_lookups = false

# ===== 性能 =====
background_gc_enabled = true
background_gc_target_interval = 2000
queue_index_embed_msgs_below = 4096

RABBITMQ_NODENAME=rabbit@myhost
RABBITMQ_MNESIA_BASE=/data/rabbitmq/mnesia
RABBITMQ_LOG_BASE=/var/log/rabbitmq
RABBITMQ_OPEN_FILES_LIMIT=65536
RABBITMQ_SERVER_START_ARGS="+S 4:4 +P 1048576"

composer require php-amqplib/php-amqplib

<?php

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 声明交换机（direct 类型）
$channel->exchange_declare('order_exchange', 'direct', false, true, false);

// 声明队列（持久化）
$channel->queue_declare('order_queue', false, true, false, false);

// 绑定队列到交换机
$channel->queue_bind('order_queue', 'order_exchange', 'order.created');

// 发送消息（持久化）
$message = new AMQPMessage(
    json_encode(['order_id' => 12345, 'user_id' => 100]),
    ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]
);

$channel->basic_publish($message, 'order_exchange', 'order.created');

$channel->close();
$connection->close();

<?php

use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 声明队列（与生产者一致）
$channel->queue_declare('order_queue', false, true, false, false);

// 手动确认
$channel->basic_consume('order_queue', '', false, false, false, false, function ($message) {
    try {
        $data = json_decode($message->body, true);
        // 处理订单业务逻辑...
        echo "Processing order: " . $data['order_id'] . PHP_EOL;
        
        // 处理成功，确认消息
        $message->ack();
    } catch (Exception $e) {
        // 处理失败，重新入队
        $message->nack(true);
    }
});

while ($channel->is_consuming()) {
    $channel->wait();
}

$channel->close();
$connection->close();