// Kafka：acks=all → 等所有副本写入才算成功 + 自动重试
Configuration config = {
    {"acks", "all"},        // ★ 关键：等所有副本确认
    {"retries", 3},         // ★ 网络抖动时自动重试
};
Producer producer(config);
producer.produce(MessageBuilder("orders").payload(body));
producer.flush();  // ★ 阻塞等待确认，不是"发完就不管"

// RabbitMQ：开启 confirm 模式，Broker 收到后回调确认
channel.confirm()
    .onSuccess([]() { /* Broker 确认收到 */ })
    .onError([](const char* msg) { /* 发送失败 */ });
channel.publish("orders", "order.created", body,
    AMQP::Mandatory,  // ★ 路由失败时回调，不会静默丢弃
    AMQP::Message().setDeliveryMode(AMQP::DeliveryMode::persistent)  // ★ 持久化
);

// RabbitMQ：声明持久化队列
channel.declareQueue("orders", AMQP::durable);  // ★ durable：队列元数据也持久化

// Kafka：关闭自动提交，处理完手动提交 offset
Configuration config = {
    {"enable.auto.commit", false},  // ★ 关键：关掉自动提交
};
Consumer consumer(config);
consumer.subscribe({"orders"});

while (true)
{
    Message msg = consumer.poll();
    if (!msg) continue;
    json order = json::parse(msg.get_payload());
    try {
        process_order(order);    // 先处理业务
        consumer.commit(msg);    // ★ 成功后才提交 offset
    } catch (const std::exception& e) {
        // 不 commit，下次会重新消费这条消息
    }
}

// RabbitMQ：手动 ACK，处理完再确认
channel.consume("orders", AMQP::noack)
    .onReceived([](const AMQP::Message& msg, uint64_t tag, bool redelivered)
    {
        try {
            process_order(order);
            channel.ack(tag);      // ★ 成功 → 确认
        } catch (...) {
            channel.nack(tag, false);  // ★ 失败 → 拒绝，转死信队列
        }
    });

// 方案一：Redis SETNX 幂等
bool first_time = redis.set("processed:" + order_id, "1",
    SetOptions{SetOptions::SetType::NX, std::chrono::seconds(86400)});
if (!first_time) return;  // ★ 已处理过，跳过
// 执行业务逻辑——只会执行一次

// Kafka：用 order_id 做 key，同一订单的消息一定进同一分区，分区内有序
producer.produce(MessageBuilder("orders")
    .key(order_id)       // ★ key 相同 → 同一 partition → 消费时按 offset 顺序
    .payload(order_data)
);

#include <cppkafka/cppkafka.h>
#include <nlohmann/json.hpp>
#include <iostream>
using namespace cppkafka;
using json = nlohmann::json;

// --- 生产者 ---
Producer producer({
    {"metadata.broker.list", "localhost:9092"}
});

json msg = {{"msg", "hello kafka"}};
producer.produce(MessageBuilder("test").payload(msg.dump()));
producer.flush();  // 确保消息发出去
std::cout << "已发送" << std::endl;

// --- 消费者 ---
Consumer consumer({
    {"metadata.broker.list", "localhost:9092"},
    {"group.id", "test-group"},
    {"auto.offset.reset", "earliest"},
});
consumer.subscribe({"test"});

while (true)
{
    Message msg = consumer.poll();
    if (msg)
    {
        json data = json::parse(msg.get_payload());
        std::cout << "收到: " << data << std::endl;
        break;  // 只收一条
    }
}

#include <amqpcpp.h>
#include <nlohmann/json.hpp>
#include <iostream>
using json = nlohmann::json;

// 假设已有 AMQP::Connection connection 和 AMQP::Channel channel(connection)

// --- 生产者 ---
channel.declareQueue("test");  // 声明队列

json payload = {{"msg", "hello rabbitmq"}};
channel.publish("", "test", payload.dump());
std::cout << "已发送" << std::endl;

// --- 消费者 ---
channel.consume("test", AMQP::noack)  // ★ noack = 手动确认
    .onReceived([](const AMQP::Message& msg, uint64_t tag, bool redelivered)
    {
        json data = json::parse(msg.body());
        std::cout << "收到: " << data << std::endl;
        channel.ack(tag);  // 手动确认
    });
std::cout << "等待消息..." << std::endl;
// channel.startConsuming();  // 启动事件循环

#include <cppkafka/cppkafka.h>
#include <nlohmann/json.hpp>
#include <iostream>
using namespace cppkafka;
using json = nlohmann::json;

int main()
{
    Configuration config = {
        {"metadata.broker.list", "localhost:9092"},

        // ★ 以下三个参数是生产环境必须配置的可靠性三件套

        {"acks", "all"},
        // acks 控制多少副本确认才算"发送成功"
        //   "0"    → 不等确认，发完就返回（最快，但可能丢消息）
        //   "1"    → 只等 Leader 副本确认（Leader 挂了仍可能丢）
        //   "all"  → 等所有 ISR 副本确认（最安全，Leader 挂了也不丢）

        {"retries", 3},
        // 网络抖动时自动重试次数。配合 retry.backoff.ms 做退避
        //   设 3 次：第 1 次失败 → 等 500ms → 第 2 次 → 等 1000ms → 第 3 次

        {"retry.backoff.ms", 500},
        // 重试间隔，避免狂打 Broker。每次重试间隔翻倍（指数退避）
    };

    Producer producer(config);

    // 构造消息体：用 nlohmann/json 序列化为 JSON 字符串
    json order = {{"order_id", "A001"}, {"amount", 99.9}};
    std::string data = order.dump();

    // ★ produce() 是异步的——消息放进本地缓冲队列，立刻返回
    producer.produce(MessageBuilder("orders").payload(data));

    // ★ flush() 是同步的——阻塞等待所有缓冲消息被 Broker 确认
    //   不调 flush，程序可能消息还没发出去就退出了
    //   生产环境通常在程序退出前或定时调用 flush
    producer.flush();

    std::cout << "发送成功" << std::endl;
    return 0;
}

#include <cppkafka/cppkafka.h>
#include <nlohmann/json.hpp>
#include <mysqlx/xdevapi.h>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
using namespace cppkafka;
using json = nlohmann::json;

Producer producer({
    {"metadata.broker.list", "localhost:9092"},
    {"acks", "all"},
    {"retries", 3},
});

std::string create_order(const std::string& user_id,
                         const std::string& product_id,
                         int quantity, double amount)
                         {
    // 生成 UUID 作为订单号
    boost::uuids::uuid uuid = boost::uuids::random_generator()();
    std::string order_id = boost::uuids::to_string(uuid);

    // 1. 写数据库——先落盘，这是"事实"
    mysqlx::Session session("localhost", 3306, "root", "123456", "shop");
    auto schema = session.getSchema("shop");
    auto orders = schema.getTable("orders");
    orders.insert("order_id", "user_id", "product_id", "quantity", "amount", "status")
          .values(order_id, user_id, product_id, quantity, amount, "created")
          .execute();
    session.close();

    // 2. 发消息——通知下游处理
    //    ★ 如果发消息失败怎么办？见下面的"本地事务表"方案
    json payload = {
        {"order_id",   order_id},
        {"user_id",    user_id},
        {"product_id", product_id},
        {"quantity",   quantity},
        {"amount",     amount},
    };
    producer.produce(MessageBuilder("order.created")
        .key(order_id)       // ★ 用 order_id 做 key，保证同一订单进同一分区
        .payload(payload.dump())
    );
    producer.flush();

    return order_id;  // 用户立即拿到订单号
}

#include <cppkafka/cppkafka.h>
#include <nlohmann/json.hpp>
#include <sw/redis++/redis++.h>
#include <mysqlx/xdevapi.h>
using namespace cppkafka;
using json = nlohmann::json;
using namespace sw::redis;

Redis redis("tcp://localhost:6379");

Consumer consumer({
    {"metadata.broker.list", "localhost:9092"},
    {"group.id", "stock-service"},   // ★ 不同的 group：各消费一份完整消息
    {"enable.auto.commit", false},    // ★ 手动提交
    {"auto.offset.reset", "earliest"},
});
consumer.subscribe({"order.created"});

void deduct_stock(const json& order)
{
    std::string order_id = order["order_id"];

    // ★ 幂等检查：Redis SETNX
    std::string key = "stock_done:" + order_id;
    bool first_time = redis.set(key, "1", SetOptions{
        SetOptions::SetType::NX,
        std::chrono::seconds(86400)
    });
    if (!first_time)
    {
        std::cout << "[库存] 订单 " << order_id << " 已扣过库存，跳过" << std::endl;
        return;
    }

    // ★ 数据库层面扣库存
    mysqlx::Session session("localhost", 3306, "root", "123456", "shop");
    auto schema = session.getSchema("shop");
    auto stock = schema.getTable("stock");
    auto result = stock.update()
        .set("quantity", mysqlx::expr("quantity - {}", order["quantity"].get<int>()))
        .where("product_id = :pid AND quantity >= :qty")
        .bind("pid", order["product_id"].get<std::string>())
        .bind("qty", order["quantity"].get<int>())
        .execute();

    if (result.getAffectedItemsCount() == 0)
    {
        std::cerr << "[库存] 商品 " << order["product_id"] << " 库存不足！" << std::endl;
        // 实际项目：发一条"库存不足"消息到死信队列
    } else
    {
        std::cout << "[库存] 订单 " << order_id << " 扣库存成功" << std::endl;
    }
    session.close();
}

while (true)
{
    Message msg = consumer.poll();
    if (!msg || msg.get_error()) continue;

    json order = json::parse(msg.get_payload());
    try {
        deduct_stock(order);
        consumer.commit(msg);  // ★ 处理成功才提交
    } catch (const std::exception& e)
    {
        std::cerr << "[库存] 处理失败: " << e.what() << "，不提交，下次重试" << std::endl;
    }
}

Consumer sms_consumer({
    {"metadata.broker.list", "localhost:9092"},
    {"group.id", "sms-service"},  // ★ 和库存服务不同的 group，各自独立消费全量消息
    {"enable.auto.commit", false},
    {"auto.offset.reset", "earliest"},
});
sms_consumer.subscribe({"order.created"});

void send_sms(const json& order)
{
    // 实际项目：调用阿里云/腾讯云短信 API
    std::cout << "[短信] 订单 " << order["order_id"]
              << " 已发送短信通知用户 " << order["user_id"] << std::endl;
}

while (true)
{
    Message msg = sms_consumer.poll();
    if (!msg || msg.get_error()) continue;

    json order = json::parse(msg.get_payload());
    try {
        send_sms(order);
        sms_consumer.commit(msg);
    } catch (const std::exception& e)
    {
        std::cerr << "[短信] 发送失败: " << e.what() << "，不提交，下次重试" << std::endl;
    }
}

std::string create_order_safe(const std::string& user_id,
                              const std::string& product_id,
                              int quantity, double amount)
                              {
    // Outbox Pattern：消息和业务数据在同一个事务里写
    boost::uuids::uuid uuid1 = boost::uuids::random_generator()();
    std::string order_id = boost::uuids::to_string(uuid1);
    boost::uuids::uuid uuid2 = boost::uuids::random_generator()();
    std::string message_id = boost::uuids::to_string(uuid2);

    mysqlx::Session session("localhost", 3306, "root", "123456", "shop");

    try {
        session.startTransaction();  // ★ 开启事务
        auto schema = session.getSchema("shop");

        // 1. 写订单
        schema.getTable("orders").insert(
            "order_id", "user_id", "product_id", "quantity", "amount", "status"
        ).values(order_id, user_id, product_id, quantity, amount, "created").execute();

        // 2. 写消息表（和订单在同一个事务里）
        json payload = {
            {"order_id",   order_id},
            {"user_id",    user_id},
            {"product_id", product_id},
            {"quantity",   quantity},
            {"amount",     amount},
        };
        schema.getTable("outbox_messages").insert(
            "message_id", "topic", "payload", "status"
        ).values(message_id, "order.created", payload.dump(), "pending").execute();

        session.commit();  // ★ 两步一起提交，要么都成功要么都失败
    } catch (...)
    {
        session.rollback();
        throw;
    }
    session.close();

    // 3. 异步：一个独立线程/进程扫描 outbox 表，把 pending 的消息发到 Kafka
    //    发送成功后更新 status='sent'
    //    ★ 即使这里挂了也没关系——重启后扫描 outbox 重新发送
    return order_id;
}

// Outbox 发送器（独立线程，定时扫描）
#include <thread>
#include <chrono>

void outbox_sender(Producer& producer)
{
    while (true)
    {
        mysqlx::Session session("localhost", 3306, "root", "123456", "shop");
        auto schema = session.getSchema("shop");
        auto outbox = schema.getTable("outbox_messages");

        // 查询 pending 消息
        auto rows = outbox.select("message_id", "topic", "payload")
            .where("status = 'pending'").limit(100).execute();

        for (auto row : rows)
        {
            std::string message_id = row[0].get<std::string>();
            std::string topic = row[1].get<std::string>();
            std::string payload = row[2].get<std::string>();

            try {
                producer.produce(MessageBuilder(topic).payload(payload));
                producer.flush();
                // 发送成功，更新状态
                outbox.update()
                    .set("status", "sent")
                    .where("message_id = :mid")
                    .bind("mid", message_id)
                    .execute();
            } catch (const std::exception& e)
            {
                std::cerr << "发送 " << message_id << " 失败: " << e.what() << "，下次重试" << std::endl;
            }
        }
        session.close();
        std::this_thread::sleep_for(std::chrono::seconds(1));  // 每秒扫描一次
    }
}

// ★ 监控积压：Kafka 消费延迟 = LogEndOffset - ConsumerOffset
#include <cppkafka/cppkafka.h>
using namespace cppkafka;

long get_lag(const std::string& topic, const std::string& group_id)
{
    Consumer consumer({
        {"metadata.broker.list", "localhost:9092"},
        {"group.id", group_id},
        {"enable.auto.commit", false},
    });

    auto partitions = consumer.get_partitions(topic);
    long total_lag = 0;

    for (const auto& p : partitions)
    {
        TopicPartition tp(topic, p);
        // LogEndOffset：分区最新 offset
        int64_t end_offset = consumer.query_offsets(tp).offset;
        // ConsumerOffset：该 group 已提交的 offset
        int64_t committed = consumer.committed(tp).get_offset();
        if (committed >= 0)
        {
            total_lag += end_offset - committed;
        }
    }
    consumer.close();
    return total_lag;
}

// 实际项目：把这个指标接入 Prometheus + Grafana 告警
// lag > 10000 时触发告警

#include <amqpcpp.h>

// AMQP::Channel channel(connection);

// 1. 声明死信队列（最终存放失败消息的地方）
channel.declareQueue("orders.dlq", AMQP::durable);

// 2. 声明主队列，绑定死信交换机
channel.declareQueue("orders", AMQP::durable | AMQP::Arguments{
    {"x-dead-letter-exchange",    ""},            // ★ 死信转发到默认交换机
    {"x-dead-letter-routing-key", "orders.dlq"}, // ★ 路由到死信队列
    {"x-max-retries", 3},                          // ★ 最多重试 3 次
});

#include <sw/redis++/redis++.h>

void consume_with_dlq(Consumer& consumer, Producer& dlq_producer,
                      std::function<void(const json&)> process_func,
                      int max_retries = 3)
                      {
    Redis redis("tcp://localhost:6379");

    while (true)
    {
        Message msg = consumer.poll();
        if (!msg || msg.get_error()) continue;

        json order = json::parse(msg.get_payload());
        std::string order_id = order["order_id"];
        std::string retry_key = "retry_count:" + order_id;

        // 读取当前重试次数
        auto val = redis.get(retry_key);
        int retry_count = val ? std::stoi(*val) : 0;

        try {
            process_func(order);
            consumer.commit(msg);
            redis.del(retry_key);  // 清理重试计数
        } catch (const std::exception& e)
        {
            retry_count++;
            redis.set(retry_key, std::to_string(retry_count),
                      std::chrono::seconds(86400));

            if (retry_count >= max_retries)
            {
                // ★ 超过最大重试次数 → 发到死信 Topic
                json dlq_msg = order;
                dlq_msg["error"] = e.what();
                dlq_msg["retry_count"] = retry_count;
                dlq_producer.produce(
                    MessageBuilder("order.created.dlq").payload(dlq_msg.dump()));
                dlq_producer.flush();
                consumer.commit(msg);  // 提交了，不再重试
                std::cerr << "[死信] 订单 " << order_id << " 转入死信队列" << std::endl;
            } else
            {
                std::cerr << "[重试 " << retry_count << "/" << max_retries
                          << "] 订单 " << order_id << ": " << e.what() << std::endl;
                // 不提交，下次重新消费
            }
        }
    }
}

#include <csignal>
#include <atomic>

std::atomic<bool> running{true};

void graceful_shutdown(int)
{
    std::cout << "收到退出信号，优雅关闭中..." << std::endl;
    running = false;
}

// 注册信号处理
std::signal(SIGTERM, graceful_shutdown);
std::signal(SIGINT, graceful_shutdown);

Consumer consumer({
    {"metadata.broker.list", "localhost:9092"},
    {"group.id", "order-processor"},
    {"enable.auto.commit", false},
});
consumer.subscribe({"orders"});

while (running)
{
    Message msg = consumer.poll(std::chrono::milliseconds(100));
    if (!msg || msg.get_error()) continue;

    json order = json::parse(msg.get_payload());
    process_order(order);
    consumer.commit(msg);  // ★ 处理完才提交
}

// ★ 退出前确保最后一条消息已提交
consumer.close();

// ★ 重置到最早的消息（从头消费）
Consumer consumer({
    {"metadata.broker.list", "localhost:9092"},
    {"group.id", "order-processor"},
    {"enable.auto.commit", false},
});

TopicPartition tp("orders", 0);
consumer.assign({tp});
consumer.seek(tp, 0);  // ★ offset=0 表示从头开始

while (true)
{
    Message msg = consumer.poll();
    if (!msg || msg.get_error()) continue;
    json data = json::parse(msg.get_payload());
    std::cout << "回溯消费: " << data << std::endl;
    break;  // 只收一条
}

// 延迟队列：消息在此停留 30 秒后自动转发到消费队列
channel.declareQueue("orders.delay", AMQP::durable | AMQP::Arguments{
    {"x-dead-letter-exchange",    ""},         // ★ 死信转发到默认交换机
    {"x-dead-letter-routing-key", "orders"},   // ★ 30 秒后转发到 orders 队列
    {"x-message-ttl", 30000},                    // ★ 消息存活 30 秒
});

// 发送时也可以单独设置 TTL
json payload = {{"order_id", "A001"}};
AMQP::Message msg(payload.dump());
msg.setDeliveryMode(AMQP::DeliveryMode::persistent);
msg.setExpiration("60000");  // ★ 这条消息 60 秒后过期转发
channel.publish("", "orders.delay", msg);

// 生产者：每个服务把日志发到 Kafka
Producer producer({
    {"metadata.broker.list", "localhost:9092"}
});

void log(const std::string& level, const std::string& service,
         const std::string& message)
         {
    // 替代 std::cout/std::log，日志直接进 Kafka
    json entry = {
        {"level",     level},
        {"service",   service},
        {"message",   message},
        {"timestamp", std::time(nullptr)},
    };
    producer.produce(MessageBuilder("app-logs").payload(entry.dump()));
}

// 用户注册后：发欢迎邮件 + 初始化积分 + 创建默认配置
// 传统做法：register() 里调三个服务 → 耦合
// EDA 做法：register() 只发一个事件，三个服务各自监听

std::string register_user(const std::string& username, const std::string& email)
{
    std::string user_id = create_user(username, email);
    // ★ 一个事件，N 个消费者各自响应
    json event = {
        {"user_id",  user_id},
        {"username", username},
        {"email",    email},
    };
    producer.produce(MessageBuilder("user.registered").payload(event.dump()));
    producer.flush();
    return user_id;
}

// 邮件服务监听 user.registered → 发欢迎邮件
// 积分服务监听 user.registered → 初始化积分账户
// 配置服务监听 user.registered → 创建默认配置
// ★ 新增"初始化推荐列表"功能？加个消费者就行，不用改注册代码

// 消费 Canal 消息，同步到 Elasticsearch
// 用 cppkafka 消费 + libcurl 调 ES API
#include <cppkafka/cppkafka.h>
#include <nlohmann/json.hpp>
#include <curl/curl.h>

Consumer consumer({
    {"metadata.broker.list", "localhost:9092"},
    {"group.id", "es-sync"},
    {"enable.auto.commit", false},
});
consumer.subscribe({"mysql.changes"});

while (true)
{
    Message msg = consumer.poll();
    if (!msg || msg.get_error()) continue;

    json change = json::parse(msg.get_payload());
    std::string type = change["type"];
    std::string id = change["data"]["id"];

    if (type == "INSERT")
    {
        // POST /products/_doc/{id}
        es_index("products", id, change["data"].dump());
    } else if (type == "UPDATE")
    {
        // POST /products/_update/{id}
        es_update("products", id, change["data"].dump());
    } else if (type == "DELETE")
    {
        // DELETE /products/_doc/{id}
        es_delete("products", change["before"]["id"]);
    }
    consumer.commit(msg);
}