RabbitMQ 技术指南（C/C++版） | 极客日志

RabbitMQ 技术指南（C/C++版） | 极客日志

+----------------+ +----------------+ +----------------+
| Producer | | RabbitMQ | | Consumer |
| (生产者) | | Broker | | (消费者) |
+-------+--------+ +-------+--------+ +-------+--------+
        |                |                |
        | 1. 发送消息    |                |
        |-------------------->|
        |                |                |
        | 2. 路由消息    |                |
        |--------+       |                |
        |        |       |                |
        |<-------+       |                |
        |                |                |
        | 3. 投递消息    |                |
        |-------------------->|
        |                |                |
        | 4. 确认消息    |                |
        |<--------------------|
        |                |                |

// 连接到指定虚拟主机
amqp_rpc_reply_t reply = amqp_login(conn, "/my_vhost", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "user", "password");

// 打开信道
amqp_channel_open(conn, 1);
amqp_rpc_reply_t reply = amqp_get_rpc_reply(conn);

#include <amqp.h>
#include <amqp_tcp_socket.h>
#include <stdio.h>
#include <stdlib.h>

void die_on_error(int x, const char* context) {
    if (x < 0) {
        fprintf(stderr, "%s: %s\n", context, amqp_error_string2(x));
        exit(1);
    }
}

void die_on_amqp_error(amqp_rpc_reply_t x, const char* context) {
    if (x.reply_type != AMQP_RESPONSE_NORMAL) {
        fprintf(stderr, "%s: AMQP error\n", context);
        exit(1);
    }
}

int main() {
    amqp_connection_state_t conn = amqp_new_connection();
    amqp_socket_t* socket = amqp_tcp_socket_new(conn);

    // 1. 连接到 RabbitMQ
    int status = amqp_socket_open(socket, "localhost", 5672);
    die_on_error(status, "opening TCP socket");

    // 2. 登录
    amqp_rpc_reply_t reply = amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest");
    die_on_amqp_error(reply, "logging in");

    // 3. 打开信道
    amqp_channel_open(conn, 1);
    reply = amqp_get_rpc_reply(conn);
    die_on_amqp_error(reply, "opening channel");

    // 4. 声明交换机
    amqp_exchange_declare(conn, 1, amqp_cstring_bytes("my_exchange"), amqp_cstring_bytes("direct"), 0, 0, 0, amqp_empty_table);
    die_on_amqp_error(amqp_get_rpc_reply(conn), "declaring exchange");

    // 5. 声明队列
    amqp_queue_declare_ok_t* queue_ok = amqp_queue_declare(conn, 1, amqp_cstring_bytes("my_queue"), 0, 0, 0, 1, amqp_empty_table);
    die_on_amqp_error(amqp_get_rpc_reply(conn), "declaring queue");

    // 6. 绑定交换机和队列
    amqp_queue_bind(conn, 1, amqp_cstring_bytes("my_queue"), amqp_cstring_bytes("my_exchange"), amqp_cstring_bytes("my_key"), amqp_empty_table);
    die_on_amqp_error(amqp_get_rpc_reply(conn), "binding queue");

    // 7. 发送消息
    const char* message = "Hello RabbitMQ!";
    amqp_basic_publish(conn, 1, amqp_cstring_bytes("my_exchange"), amqp_cstring_bytes("my_key"), 0, 0, NULL, amqp_cstring_bytes(message));
    printf("Sent message: %s\n", message);

    // 8. 清理资源
    amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);
    amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
    amqp_destroy_connection(conn);
    return 0;
}

// 声明 Direct 交换机
amqp_exchange_declare(conn, 1, amqp_cstring_bytes("direct_exchange"), amqp_cstring_bytes("direct"), 0, 0, 0, amqp_empty_table);
// 绑定队列到指定路由键
amqp_queue_bind(conn, 1, amqp_cstring_bytes("queue1"), amqp_cstring_bytes("direct_exchange"), amqp_cstring_bytes("error"), amqp_empty_table);
amqp_queue_bind(conn, 1, amqp_cstring_bytes("queue2"), amqp_cstring_bytes("direct_exchange"), amqp_cstring_bytes("info"), amqp_empty_table);
// 发送消息到指定路由键
amqp_basic_publish(conn, 1, amqp_cstring_bytes("direct_exchange"), amqp_cstring_bytes("error"), 0, 0, NULL, amqp_cstring_bytes("Error message"));

// 声明 Fanout 交换机
amqp_exchange_declare(conn, 1, amqp_cstring_bytes("fanout_exchange"), amqp_cstring_bytes("fanout"), 0, 0, 0, amqp_empty_table);
// 绑定多个队列
amqp_queue_bind(conn, 1, amqp_cstring_bytes("queue1"), amqp_cstring_bytes("fanout_exchange"), amqp_empty_bytes, amqp_empty_table);
amqp_queue_bind(conn, 1, amqp_cstring_bytes("queue2"), amqp_cstring_bytes("fanout_exchange"), amqp_empty_bytes, amqp_empty_table);
// 发送广播消息
amqp_basic_publish(conn, 1, amqp_cstring_bytes("fanout_exchange"), amqp_empty_bytes, 0, 0, NULL, amqp_cstring_bytes("Broadcast message"));

// 声明 Topic 交换机
amqp_exchange_declare(conn, 1, amqp_cstring_bytes("topic_exchange"), amqp_cstring_bytes("topic"), 0, 0, 0, amqp_empty_table);
// 绑定队列到主题
amqp_queue_bind(conn, 1, amqp_cstring_bytes("queue1"), amqp_cstring_bytes("topic_exchange"), amqp_cstring_bytes("user.*"), amqp_empty_table);
// 匹配 user.create, user.update
amqp_queue_bind(conn, 1, amqp_cstring_bytes("queue2"), amqp_cstring_bytes("topic_exchange"), amqp_cstring_bytes("order.#"), amqp_empty_table);
// 匹配 order.create, order.pay.success
// 发送消息到指定主题
amqp_basic_publish(conn, 1, amqp_cstring_bytes("topic_exchange"), amqp_cstring_bytes("user.create"), 0, 0, NULL, amqp_cstring_bytes("User created"));

// 声明 Headers 交换机
amqp_exchange_declare(conn, 1, amqp_cstring_bytes("headers_exchange"), amqp_cstring_bytes("headers"), 0, 0, 0, amqp_empty_table);
// 设置匹配头信息
amqp_table_t headers;
amqp_table_entry_t entry;
entry.key = amqp_cstring_bytes("type");
entry.value.kind = AMQP_FIELD_KIND_UTF8;
entry.value.value.utf8 = amqp_cstring_bytes("email");
headers.num_entries = 1;
headers.entries = &entry;
// 绑定队列
amqp_queue_bind(conn, 1, amqp_cstring_bytes("queue1"), amqp_cstring_bytes("headers_exchange"), amqp_empty_bytes, headers);
// 发送带头部的消息
amqp_basic_properties_t props;
props._flags = AMQP_BASIC_HEADERS_FLAG;
props.headers = headers;
amqp_basic_publish(conn, 1, amqp_cstring_bytes("headers_exchange"), amqp_empty_bytes, 0, 0, &props, amqp_cstring_bytes("Email message"));

// 订单服务 - 发送消息
void create_order(const char* order_id, int product_id, int quantity) {
    // 保存订单到数据库
    save_order_to_db(order_id, product_id, quantity);
    // 发送消息到库存系统
    char message[256];
    snprintf(message, sizeof(message), "{\"order_id\":\"%s\",\"product_id\":%d,\"quantity\":%d}", order_id, product_id, quantity);
    amqp_basic_publish(conn, 1, amqp_cstring_bytes("order_exchange"), amqp_cstring_bytes("inventory.update"), 0, 0, NULL, amqp_cstring_bytes(message));
}

// 库存服务 - 消费消息
void consume_inventory_updates() {
    amqp_basic_consume(conn, 1, amqp_cstring_bytes("inventory_queue"), amqp_empty_bytes, 0, 0, 0, amqp_empty_table);
    while (1) {
        amqp_envelope_t envelope;
        amqp_consume_message(conn, &envelope, NULL, 0);
        // 解析消息并更新库存
        update_inventory((char*)envelope.message.body.bytes);
        // 手动确认
        amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
        amqp_destroy_envelope(&envelope);
    }
}

// 用户注册服务
void register_user(const char* username, const char* email, const char* phone) {
    // 保存用户信息
    save_user(username, email, phone);
    // 发送邮件通知消息
    char email_msg[256];
    snprintf(email_msg, sizeof(email_msg), "{\"email\":\"%s\",\"username\":\"%s\"}", email, username);
    amqp_basic_publish(conn, 1, amqp_cstring_bytes("notification_exchange"), amqp_cstring_bytes("email.send"), 0, 0, NULL, amqp_cstring_bytes(email_msg));
    // 发送短信通知消息
    char sms_msg[256];
    snprintf(sms_msg, sizeof(sms_msg), "{\"phone\":\"%s\",\"username\":\"%s\"}", phone, username);
    amqp_basic_publish(conn, 1, amqp_cstring_bytes("notification_exchange"), amqp_cstring_bytes("sms.send"), 0, 0, NULL, amqp_cstring_bytes(sms_msg));
}

// 秒杀接口
void seckill_handler(const char* user_id, const char* product_id) {
    // 将请求放入消息队列
    char message[256];
    snprintf(message, sizeof(message), "{\"user_id\":\"%s\",\"product_id\":\"%s\"}", user_id, product_id);
    // 设置消息持久化
    amqp_basic_properties_t props;
    props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG;
    props.delivery_mode = 2; // 持久化
    amqp_basic_publish(conn, 1, amqp_cstring_bytes("seckill_exchange"), amqp_cstring_bytes("order.create"), 0, 0, &props, amqp_cstring_bytes(message));
    printf("Seckill request received for user %s, product %s\n", user_id, product_id);
}

// 订单处理消费者（控制并发数）
void process_seckill_orders() {
    // 设置 QoS，控制消费速度
    amqp_basic_qos(conn, 1, 10, 0); // 每次最多处理 10 条消息
    amqp_basic_consume(conn, 1, amqp_cstring_bytes("seckill_queue"), amqp_empty_bytes, 0, 0, 0, amqp_empty_table);
    while (1) {
        amqp_envelope_t envelope;
        amqp_consume_message(conn, &envelope, NULL, 0);
        // 处理秒杀订单
        process_seckill_order((char*)envelope.message.body.bytes);
        amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
        amqp_destroy_envelope(&envelope);
    }
}

+----------------+ +----------------+ +----------------+
| Node 1 | | Node 2 | | Node 3 |
| (Disk Node) | | (RAM Node) | | (RAM Node) |
+-------+--------+ +-------+--------+ +-------+--------+
        |                |                |
        |                |                |
+---------------------+---------------------+
| +---------+---------+ |
| +-----+-----+ +-----+-----+ |
| Load | | Client | |
| Balancer |<------| Applications | |
+-----------+ +-----------+ |

# 配置镜像队列策略（所有队列都镜像到所有节点）
rabbitmqctl set_policy ha-all "^"'{"ha-mode":"all"}'
# 配置镜像队列策略（指定队列镜像到 2 个节点）
rabbitmqctl set_policy ha-two "^important-"'{"ha-mode":"exactly","ha-params":2}'

#include <vector>
#include <string>

// 集群节点配置
std::vector<std::string> cluster_nodes = {"192.168.1.10:5672", "192.168.1.11:5672", "192.168.1.12:5672"};

amqp_connection_state_t connect_to_cluster() {
    amqp_connection_state_t conn = amqp_new_connection();
    // 尝试连接到集群中的节点
    for (const auto& node : cluster_nodes) {
        size_t colon_pos = node.find(':');
        std::string host = node.substr(0, colon_pos);
        int port = std::stoi(node.substr(colon_pos + 1));
        amqp_socket_t* socket = amqp_tcp_socket_new(conn);
        if (!socket) continue;
        int status = amqp_socket_open(socket, host.c_str(), port);
        if (status == 0) {
            // 连接成功，进行登录
            amqp_rpc_reply_t reply = amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest");
            if (reply.reply_type == AMQP_RESPONSE_NORMAL) {
                amqp_channel_open(conn, 1);
                amqp_get_rpc_reply(conn);
                return conn;
            }
        }
    }
    // 所有节点连接失败
    amqp_destroy_connection(conn);
    return NULL;
}

// 连接参数配置
struct RabbitMQConfig {
    std::vector<std::string> nodes;
    std::string username;
    std::string password;
    std::string vhost;
    int heartbeat_interval;
    int connection_timeout;
    int retry_attempts;
    int retry_delay;
};

// 重连机制
void reconnect_loop(amqp_connection_state_t* conn, const RabbitMQConfig& config) {
    int attempt = 0;
    while (attempt < config.retry_attempts) {
        *conn = connect_to_cluster(config);
        if (*conn != NULL) {
            printf("Connected to RabbitMQ cluster successfully\n");
            return;
        }
        attempt++;
        printf("Connection attempt %d failed, retrying in %d seconds...\n", attempt, config.retry_delay);
        sleep(config.retry_delay);
    }
    fprintf(stderr, "Failed to connect to RabbitMQ cluster after %d attempts\n", config.retry_attempts);
    exit(1);
}

// 心跳检测
void heartbeat_monitor(amqp_connection_state_t conn, int interval) {
    while (1) {
        amqp_frame_t frame;
        int res = amqp_simple_wait_frame(conn, &frame);
        if (res != AMQP_STATUS_OK) {
            fprintf(stderr, "Heartbeat failed, connection lost\n");
            // 触发重连逻辑
            return;
        }
        // 处理心跳帧
        if (frame.frame_type == AMQP_FRAME_HEARTBEAT) {
            amqp_send_heartbeat(conn);
        }
    }
}

// 连接池实现
class ConnectionPool {
private:
    std::vector<amqp_connection_state_t> connections;
    std::mutex pool_mutex;
    int max_connections;
public:
    ConnectionPool(int max_conn) : max_connections(max_conn) {
        // 初始化连接池
        for (int i = 0; i < max_conn; i++) {
            amqp_connection_state_t conn = create_connection();
            if (conn) connections.push_back(conn);
        }
    }

    amqp_connection_state_t get_connection() {
        std::lock_guard<std::mutex> lock(pool_mutex);
        if (!connections.empty()) {
            amqp_connection_state_t conn = connections.back();
            connections.pop_back();
            return conn;
        }
        // 连接池为空，创建新连接
        return create_connection();
    }

    void release_connection(amqp_connection_state_t conn) {
        std::lock_guard<std::mutex> lock(pool_mutex);
        if (connections.size() < max_connections) {
            connections.push_back(conn);
        } else {
            // 超过最大连接数，关闭连接
            close_connection(conn);
        }
    }

private:
    amqp_connection_state_t create_connection() {
        amqp_connection_state_t conn = amqp_new_connection();
        amqp_socket_t* socket = amqp_tcp_socket_new(conn);
        if (amqp_socket_open(socket, "localhost", 5672) != 0) {
            amqp_destroy_connection(conn);
            return NULL;
        }
        amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest");
        amqp_channel_open(conn, 1);
        amqp_get_rpc_reply(conn);
        return conn;
    }

    void close_connection(amqp_connection_state_t conn) {
        amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);
        amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
        amqp_destroy_connection(conn);
    }
};

// 批量发送消息
void batch_publish(amqp_connection_state_t conn, const std::vector<std::string>& messages) {
    // 开始事务
    amqp_tx_select(conn, 1);
    for (const auto& msg : messages) {
        amqp_basic_publish(conn, 1, amqp_cstring_bytes("batch_exchange"), amqp_cstring_bytes("batch.key"), 0, 0, NULL, amqp_cstring_bytes(msg.c_str()));
    }
    // 提交事务
    amqp_tx_commit(conn, 1);
}

// 批量消费消息
void batch_consume(amqp_connection_state_t conn, const char* queue_name, int batch_size) {
    std::vector<amqp_envelope_t> batch;
    batch.reserve(batch_size);
    while (batch.size() < batch_size) {
        amqp_envelope_t envelope;
        amqp_rpc_reply_t reply = amqp_consume_message(conn, &envelope, NULL, 100);
        if (reply.reply_type == AMQP_RESPONSE_NORMAL) {
            batch.push_back(envelope);
        } else if (reply.library_error == AMQP_STATUS_TIMEOUT) {
            // 超时，处理已有消息
            break;
        }
    }
    // 批量处理消息
    for (auto& envelope : batch) {
        process_message((char*)envelope.message.body.bytes);
        amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
        amqp_destroy_envelope(&envelope);
    }
}

// QoS 配置
void configure_qos(amqp_connection_state_t conn, int prefetch_count, int prefetch_size) {
    // 设置信道级别的 QoS
    amqp_basic_qos(conn, 1, prefetch_size, prefetch_count);
    // 或者设置连接级别的 QoS
    // amqp_basic_qos(conn, 0, prefetch_size, prefetch_count);
}

// 消费者并发处理
void start_consumer_workers(int worker_count) {
    for (int i = 0; i < worker_count; i++) {
        pthread_t thread;
        pthread_create(&thread, NULL, consumer_worker, NULL);
    }
}

void* consumer_worker(void* arg) {
    amqp_connection_state_t conn = create_connection();
    // 设置每个消费者的 QoS
    amqp_basic_qos(conn, 1, 0, 10); // 每个消费者预取 10 条消息
    amqp_basic_consume(conn, 1, amqp_cstring_bytes("task_queue"), amqp_empty_bytes, 0, 0, 0, amqp_empty_table);
    while (1) {
        amqp_envelope_t envelope;
        amqp_consume_message(conn, &envelope, NULL, 0);
        process_task((char*)envelope.message.body.bytes);
        amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
        amqp_destroy_envelope(&envelope);
    }
    return NULL;
}

// 持久化配置
void configure_persistence(amqp_connection_state_t conn) {
    // 1. 声明持久化交换机
    amqp_exchange_declare(conn, 1, amqp_cstring_bytes("persistent_exchange"), amqp_cstring_bytes("direct"), 1, 0, 0, amqp_empty_table);
    // 2. 声明持久化队列
    amqp_queue_declare_ok_t* queue_ok = amqp_queue_declare(conn, 1, amqp_cstring_bytes("persistent_queue"), 1, 0, 0, 0, amqp_empty_table);
    // 3. 绑定队列
    amqp_queue_bind(conn, 1, amqp_cstring_bytes("persistent_queue"), amqp_cstring_bytes("persistent_exchange"), amqp_cstring_bytes("persistent.key"), amqp_empty_table);
    // 4. 发送持久化消息
    amqp_basic_properties_t props;
    props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG;
    props.delivery_mode = 2; // 持久化
    const char* message = "Persistent message";
    amqp_basic_publish(conn, 1, amqp_cstring_bytes("persistent_exchange"), amqp_cstring_bytes("persistent.key"), 0, 0, &props, amqp_cstring_bytes(message));
}

// 完整的可靠性配置
void reliable_message_delivery() {
    amqp_connection_state_t conn = create_connection();
    // 1. 声明持久化交换机和队列
    amqp_exchange_declare(conn, 1, amqp_cstring_bytes("reliable_exchange"), amqp_cstring_bytes("direct"), 1, 0, 0, amqp_empty_table);
    amqp_queue_declare_ok_t* queue_ok = amqp_queue_declare(conn, 1, amqp_cstring_bytes("reliable_queue"), 1, 0, 0, 0, amqp_empty_table);
    amqp_queue_bind(conn, 1, amqp_cstring_bytes("reliable_queue"), amqp_cstring_bytes("reliable_exchange"), amqp_cstring_bytes("reliable.key"), amqp_empty_table);
    // 2. 启用发布确认
    amqp_confirm_select(conn, 1);
    // 3. 发送持久化消息
    amqp_basic_properties_t props;
    props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG;
    props.delivery_mode = 2;
    const char* message = "Reliable message";
    amqp_basic_publish(conn, 1, amqp_cstring_bytes("reliable_exchange"), amqp_cstring_bytes("reliable.key"), 0, 0, &props, amqp_cstring_bytes(message));
    // 4. 等待发布确认
    amqp_frame_t frame;
    int result = amqp_simple_wait_frame(conn, &frame);
    if (result == AMQP_STATUS_OK && frame.payload.method.id == AMQP_BASIC_ACK_METHOD) {
        printf("Message published successfully\n");
    } else {
        fprintf(stderr, "Message publish failed\n");
    }
    // 5. 消费者手动确认
    amqp_basic_consume(conn, 1, amqp_cstring_bytes("reliable_queue"), amqp_empty_bytes, 0, 0, 0, amqp_empty_table);
    amqp_envelope_t envelope;
    amqp_consume_message(conn, &envelope, NULL, 0);
    process_message((char*)envelope.message.body.bytes);
    // 手动确认消息
    amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
    amqp_destroy_envelope(&envelope);
}

#include <unordered_set>
#include <mutex>

class IdempotentConsumer {
private:
    std::unordered_set<std::string> processed_messages;
    std::mutex message_mutex;
public:
    void consume_message(amqp_envelope_t& envelope) {
        // 1. 获取消息 ID
        std::string message_id;
        if (envelope.message.properties._flags & AMQP_BASIC_MESSAGE_ID_FLAG) {
            message_id = std::string((char*)envelope.message.properties.message_id.bytes, envelope.message.properties.message_id.len);
        } else {
            // 如果没有消息 ID，生成一个
            message_id = generate_unique_id();
        }
        // 2. 检查消息是否已处理
        std::lock_guard<std::mutex> lock(message_mutex);
        if (processed_messages.find(message_id) != processed_messages.end()) {
            printf("Duplicate message detected: %s\n", message_id.c_str());
            amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
            return;
        }
        try {
            // 3. 处理消息
            process_business_logic((char*)envelope.message.body.bytes);
            // 4. 标记消息为已处理
            processed_messages.insert(message_id);
            // 5. 确认消息
            amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
        } catch (const std::exception& e) {
            fprintf(stderr, "Message processing failed: %s\n", e.what());
            // 6. 消息处理失败，不确认，让消息重新投递
        }
    }

private:
    std::string generate_unique_id() {
        // 生成唯一 ID 的实现
        static int counter = 0;
        char id[64];
        snprintf(id, sizeof(id), "msg_%d_%ld", counter++, time(NULL));
        return std::string(id);
    }
};

// 单队列单消费者保证顺序
void sequential_message_processing() {
    amqp_connection_state_t conn = create_connection();
    // 1. 使用单个队列
    amqp_queue_declare_ok_t* queue_ok = amqp_queue_declare(conn, 1, amqp_cstring_bytes("sequential_queue"), 1, 0, 0, 0, amqp_empty_table);
    // 2. 单个消费者
    amqp_basic_consume(conn, 1, amqp_cstring_bytes("sequential_queue"), amqp_empty_bytes, 0, 0, 0, amqp_empty_table);
    // 3. 单线程处理
    while (1) {
        amqp_envelope_t envelope;
        amqp_consume_message(conn, &envelope, NULL, 0);
        // 按顺序处理消息
        process_message_in_order((char*)envelope.message.body.bytes);
        amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
        amqp_destroy_envelope(&envelope);
    }
}

// 按业务键分区保证顺序
void partitioned_message_processing() {
    // 为不同的业务键创建不同的队列
    const char* business_keys[] = {"user1", "user2", "user3", NULL};
    for (int i = 0; business_keys[i] != NULL; i++) {
        char queue_name[64];
        snprintf(queue_name, sizeof(queue_name), "partition_queue_%s", business_keys[i]);
        amqp_queue_declare(conn, 1, amqp_cstring_bytes(queue_name), 1, 0, 0, 0, amqp_empty_table);
        // 每个队列一个消费者
        pthread_t thread;
        pthread_create(&thread, NULL, partition_consumer, (void*)queue_name);
    }
}

void* partition_consumer(void* queue_name_ptr) {
    const char* queue_name = (const char*)queue_name_ptr;
    amqp_connection_state_t conn = create_connection();
    amqp_basic_consume(conn, 1, amqp_cstring_bytes(queue_name), amqp_empty_bytes, 0, 0, 0, amqp_empty_table);
    while (1) {
        amqp_envelope_t envelope;
        amqp_consume_message(conn, &envelope, NULL, 0);
        process_message((char*)envelope.message.body.bytes);
        amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
        amqp_destroy_envelope(&envelope);
    }
    return NULL;
}

// 性能优化综合配置
void optimize_performance() {
    amqp_connection_state_t conn = create_connection();
    // 1. 连接优化
    amqp_socket_set_timeout(socket, 30); // 设置超时
    amqp_set_heartbeat(conn, 60); // 设置心跳
    // 2. QoS 优化
    amqp_basic_qos(conn, 1, 0, 100); // 预取 100 条消息
    // 3. 启用消息压缩
    amqp_compression_t compression = amqp_compression_gzip;
    amqp_socket_set_compression(socket, compression);
    // 4. 批量操作
    std::vector<std::string> messages;
    for (int i = 0; i < 1000; i++) {
        messages.push_back("Batch message " + std::to_string(i));
    }
    batch_publish(conn, messages);
    // 5. 异步处理
    start_consumer_workers(10); // 启动 10 个消费者线程
}

// 异步消费者模式
void async_consumer_example() {
    amqp_connection_state_t conn = create_connection();
    // 设置异步回调
    amqp_set_message_handler(conn, message_handler);
    amqp_set_error_handler(conn, error_handler);
    // 开始异步消费
    amqp_basic_consume(conn, 1, amqp_cstring_bytes("async_queue"), amqp_empty_bytes, 0, 1, 0, amqp_empty_table);
    // 事件循环
    while (1) {
        amqp_maybe_release_buffers(conn);
        amqp_wait_for_event(conn, 100);
    }
}

void message_handler(amqp_envelope_t* envelope) {
    // 异步处理消息
    process_message_async((char*)envelope->message.body.bytes);
    amqp_basic_ack(envelope->channel->connection, envelope->channel->channel_id, envelope->delivery_tag, 0);
}

# 1. 设置内存阈值
rabbitmqctl set_vm_memory_high_watermark 0.4
# 2. 启用磁盘持久化
rabbitmqctl set_policy persistence "^"'{"delivery-mode":2}'
# 3. 设置队列最大长度
rabbitmqctl set_policy max-length "^"'{"x-max-length":10000}'
# 4. 启用懒惰队列
rabbitmqctl set_policy lazy-queue "^lazy-"'{"x-queue-mode":"lazy"}'

// 客户端配置优化
void memory_optimization() {
    amqp_connection_state_t conn = amqp_new_connection();
    // 1. 限制消息大小
    const int MAX_MESSAGE_SIZE = 1024 * 1024; // 1MB
    // 2. 使用内存池
    amqp_pool_t* pool = amqp_pool_new(1024 * 1024); // 1MB 内存池
    // 3. 及时释放内存
    amqp_maybe_release_buffers(conn);
    // 4. 限制预取大小
    amqp_basic_qos(conn, 1, 1024 * 1024, 10); // 预取 10 条，每条最大 1MB
    // 5. 批量释放
    std::vector<amqp_envelope_t> messages;
    // 处理消息...
    for (auto& env : messages) {
        amqp_destroy_envelope(&env);
    }
    messages.clear();
}