c++ Kafka的流处理API进行行情数据处理代码

以下是使用C++编写Kafka流处理程序，从Kafka主题中读取行情数据，并进行处理的示例代码：

#include <iostream>
#include <string>
#include <librdkafka/rdkafkacpp.h>

class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
 public:
  void dr_cb(RdKafka::Message &message) {
    std::cout << "Message delivery for (" << message.len() << " bytes): " <<
                 message.errstr() << std::endl;
  }
};

class ExampleEventCb : public RdKafka::EventCb {
 public:
  void event_cb(RdKafka::Event &event) {
    switch (event.type()) {
      case RdKafka::Event::EVENT_ERROR:
        std::cerr << "ERROR (" << RdKafka::err2str(event.err()) << "): " <<
                     event.str() << std::endl;
        if (event.err() == RdKafka::ERR__ALL_BROKERS_DOWN)
          exit(1);
        break;

      case RdKafka::Event::EVENT_STATS:
        std::cerr << "\"STATS\": " << event.str() << std::endl;
        break;

      case RdKafka::Event::EVENT_LOG:
        fprintf(stderr, "LOG-%i-%s: %s\n",
                event.severity(), event.fac().c_str(), event.str().c_str());
        break;

      default:
        std::cerr << "EVENT " << event.type() <<
                     " (" << RdKafka::err2str(event.err()) << "): " <<
                     event.str() << std::endl;
        break;
    }
  }
};

int main() {
  std::string brokers = "localhost:9092";
  std::string errstr;

  RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  conf->set("bootstrap.servers", brokers, errstr);

  ExampleDeliveryReportCb ex_dr_cb;
  conf->set("dr_cb", &ex_dr_cb, errstr);

  ExampleEventCb ex_event_cb;
  conf->set("event_cb", &ex_event_cb, errstr);

  RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);

  std::string topic_str = "test";
  RdKafka::Topic *topic = RdKafka::Topic::create(producer, topic_str,
                                                  NULL, errstr);

  for (int i = 0; i < 10; i++) {
    std::string payload = "Message " + std::to_string(i);
    RdKafka::ErrorCode resp =
        producer->produce(topic, RdKafka::Topic::PARTITION_UA,
                          RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
                          const_cast<char *>(payload.c_str()), payload.size(),
                          NULL, NULL);
    if (resp != RdKafka::ERR_NO_ERROR)
      std::cerr << "% Produce failed: " <<
                   RdKafka::err2str(resp) << std::endl;
    else
      std::cerr << "% Produced message (" << payload.size() << " bytes)" <<
                   std::endl;
    producer->poll(0);
  }

  delete topic;
  delete producer;
  return 0;
}

此代码展示了使用librdkafka库编写Kafka生产者程序的基本步骤，包括创建Kafka生产者对象、创建Kafka主题、将行情数据发送到Kafka主题中等。需要注意的是，此代码仅为示例代码，需要根据实际情况进行修改和优化。

以下是使用C++编写Kafka流处理程序的示例代码：

#include <iostream>
#include <string>
#include <librdkafka/rdkafkacpp.h>

class ExampleRebalanceCb : public RdKafka::RebalanceCb {
 public:
  void rebalance_cb(RdKafka::KafkaConsumer *consumer,
                     RdKafka::ErrorCode err,
                     std::vector<RdKafka::TopicPartition *> &partitions) {
    std::cout << "RebalanceCb: " << RdKafka::err2str(err) << std::endl;
    if (err == RdKafka::ERR__ASSIGN_PARTITIONS) {
      consumer->assign(partitions);
    } else {
      consumer->unassign();
    }
  }
};

class ExampleConsumeCb : public RdKafka::ConsumeCb {
 public:
  void consume_cb(RdKafka::Message &message, void *opaque) {
    if (message.err() == RdKafka::ERR_NO_ERROR) {
      std::string payload(static_cast<const char *>(message.payload()),
                          message.len());
      std::cout << "Message: " << payload << std::endl;
    } else {
      std::cerr << "Consume failed: " << message.errstr() << std::endl;
    }
  }
};

int main() {
  std::string brokers = "localhost:9092";
  std::string group = "test_group";
  std::string topic_str = "test";
  std::string errstr;

  RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  conf->set("bootstrap.servers", brokers, errstr);
  conf->set("group.id", group, errstr);
  conf->set("enable.auto.commit", "false", errstr);

  ExampleRebalanceCb ex_rebalance_cb;
  conf->set("rebalance_cb", &ex_rebalance_cb, errstr);

  RdKafka::KafkaConsumer *consumer =
      RdKafka::KafkaConsumer::create(conf, errstr);

  RdKafka::TopicPartition *part =
      RdKafka::TopicPartition::create(topic_str, 0, RdKafka::Topic::OFFSET_BEGINNING);
  std::vector<RdKafka::TopicPartition *> parts = {part};

  ExampleConsumeCb ex_consume_cb;
  int ret = consumer->subscribe(parts);
  while (true) {
    RdKafka::Message *message = consumer->consume(1000);
    ex_consume_cb.consume_cb(*message, NULL);
    consumer->commitSync(message);
    delete message;
  }

  delete consumer;
  return 0;
}

此代码展示了使用librdkafka库编写Kafka流处理程序的基本步骤，包括创建Kafka消费者对象、订阅Kafka主题、从Kafka主题中读取消息等。需要注意的是，此代码仅为示例代码，需要根据实际情况进行修改和优化。