C++操作Kafka

脚本测试

1. 部署zookeeper
2. 启动broker 1
3. 启动broker 2
4. 创建主题、查看主题

   1 2	kafka-topics.sh --create --zookeeper 192.168.31.249:2181 -replication-factor 2 --partitions 2 --topic kafka-2 kafka-toplics.sh --describe --zookeeper 192.168.31.249:2181 --topic kafka-2

5. 测试集群

   1 2 3 4 5

   ## 生产者 kafaka-console-producer.sh --broker-list 192.168.31.249:9092 --topic kafka-2 ## 两个消费者 kafaka-console-consumer.sh --bootstrap-server 192.168.31.249:9092 --topic kafka-2 --group 0 --from-beginning kafaka-console-consumer.sh --bootstrap-server 192.168.31.249:9092 --topic kafka-2 --group 0 --from-beginning

C++开发

生产者

0. 定义分区策略

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27

   // 可以自己定义分区策略 class HashPartitionerCb : public RdKafka::PartitionerCb { public: int32_t partitioner_cb(const RdKafka::Topic *topic, const std::string *key, int32_t partition_cnt, void *msg_opaque) { char msg[128] = {0}; int32_t partition_id = generate_hash(key->c_str(), key->size()) % partition_cnt; // [topic][key][partition_cnt][partition_id] // :[test][6419][2][1] sprintf(msg, "HashPartitionerCb:topic:[%s], key:[%s]partition_cnt:[%d], " "partition_id:[%d]", topic->name().c_str(), key->c_str(), partition_cnt, partition_id); // std::cout << msg << std::endl; return partition_id; } private: static inline unsigned int generate_hash(const char *str, size_t len) { unsigned int hash = 5381; for (size_t i = 0; i < len; i++) hash = ((hash << 5) + hash) + str[i]; return hash; } };

1. 配置

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35

   // 创建Kafka Conf对象 m_config = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL); if (m_config == NULL) { std::cout << "Create RdKafka Conf failed." << std::endl; } // 创建Topic Conf对象 m_topicConfig = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC); if (m_topicConfig == NULL) { std::cout << "Create RdKafka Topic Conf failed." << std::endl; } // 设置Broker属性 RdKafka::Conf::ConfResult errCode; m_dr_cb = new ProducerDeliveryReportCb; // 创建一个回调类 std::string errorStr; // 传递情况报告 errCode = m_config->set("dr_cb", m_dr_cb, errorStr); // 异步方式发送数据 if (errCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed:" << errorStr << std::endl; } m_event_cb = new ProducerEventCb; errCode = m_config->set("event_cb", m_event_cb, errorStr); if (errCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed:" << errorStr << std::endl; } // 可以设置自己的分区策略 m_partitioner_cb = new HashPartitionerCb; errCode = m_topicConfig->set("partitioner_cb", m_partitioner_cb, errorStr); if (errCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed:" << errorStr << std::endl; } // 设置broker的地址 errCode = m_config->set("bootstrap.servers", m_brokers, errorStr); if (errCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed:" << errorStr << std::endl; }

2. 创建生产者

• kafka提供的异步的生产接口，异步和同步的消费接口，没有同步的生产接口

  1 2 3 4 5 6 7 8 9 10

  // 创建Producer，可以发布不同的主题 m_producer = RdKafka::Producer::create(m_config, errorStr); if (m_producer == NULL) { std::cout << "Create Producer failed:" << errorStr << std::endl; } // 创建Topic对象，不同的主题可以传入不同的m_topicStr，也就是说可以使用不同的分区策略 m_topic = RdKafka::Topic::create(m_producer, m_topicStr, m_topicConfig, errorStr); if (m_topic == NULL) { std::cout << "Create Topic failed:" << errorStr << std::endl; }

3. 发送消息

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

   // API形式1 RdKafka::ErrorCode errorCode = m_producer->produce( m_topic, // 指定发送到哪个主题 RdKafka::Topic::PARTITION_UA, // 指定分区，如果为PARTITION_UA则通过 // partitioner_cb的回调选择合适的分区 RdKafka::Producer::RK_MSG_COPY, // 消息拷贝 payload, // 消息本身 len, // 消息长度 &key, // 消息key NULL // an optional application-provided per-message opaque pointer // that will be provided in the message delivery callback to let // the application reference a specific message ); // API形式2 RdKafka::ProducerImpl::produce(const std::string topic_name, int32_t partition, int msgflags, void *payload, size_t len, const void *key, size_t key_len, int64_t timestamp, RdKafka::Headers *headers, void *msg_opaque ) // 假如不设置timestamp或者使用API形式1，那么每个消息仍然是有时间戳的，这是kafka协议规定的，c++kafka客户端内部会自己添加时间戳。 // 时间戳用来干嘛？kafka可以设置日志什么时候删除，比如保存最近5天的，这时候就需要时间戳 m_producer->poll(0); // 必须要poll一下，不然设置的回调不会触发

• 重要参数
  1. batch.size
  2. linger.ms

消费者

1. 配置

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30

   std::string errorStr; RdKafka::Conf::ConfResult errorCode; m_config = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL); m_event_cb = new ConsumerEventCb; errorCode = m_config->set("event_cb", m_event_cb, errorStr); if(errorCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed: " << errorStr << std::endl; } m_rebalance_cb = new ConsumerRebalanceCb; errorCode = m_config->set("rebalance_cb", m_rebalance_cb, errorStr); if(errorCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed: " << errorStr << std::endl; } errorCode = m_config->set("group.id", m_groupID, errorStr); if(errorCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed: " << errorStr << std::endl; } errorCode = m_config->set("bootstrap.servers", m_brokers, errorStr); if(errorCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed: " << errorStr << std::endl; } // partition.assignment.strategy range,roundrobin errorCode = m_config->set("partition.assignment.strategy", "range", errorStr); if (errorCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed: " << errorStr << std::endl; }

2. 订阅主题与分区

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

   m_topicConfig = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC); // 获取最新的消息数据 errorCode = m_topicConfig->set("auto.offset.reset", "latest", errorStr); if (errorCode != RdKafka::Conf::CONF_OK) { std::cout << "Topic Conf set failed: " << errorStr << std::endl; } errorCode = m_config->set("default_topic_conf", m_topicConfig, errorStr); if (errorCode != RdKafka::Conf::CONF_OK) { std::cout << "Conf set failed: " << errorStr << std::endl; } m_consumer = RdKafka::KafkaConsumer::create(m_config, errorStr); if (m_consumer == NULL) { std::cout << "Create KafkaConsumer failed: " << errorStr << std::endl; } std::cout << "Created consumer " << m_consumer->name() << std::endl; // 订阅主题, 可以订阅多个主题，所以传入vector RdKafka::ErrorCode errorCode = m_consumer->subscribe(m_topicVector);

3. 消息消费

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27

   void KafkaConsumer::pullMessage() { // 订阅Topic RdKafka::ErrorCode errorCode = m_consumer->subscribe(m_topicVector); if (errorCode != RdKafka::ERR_NO_ERROR) { std::cout << "subscribe failed: " << RdKafka::err2str(errorCode) << std::endl; } // 消费消息 while (true) { RdKafka::Message *msg = m_consumer->consume(1000); // 1s超时 msg_consume(msg, NULL); delete msg; } } void msg_consume(RdKafka::Message *msg, void *opaque) { switch (msg->err()) { case RdKafka::ERR__TIMED_OUT: std::cerr << "Consumer error: " << msg->errstr() << std::endl; // 超时 break; case RdKafka::ERR_NO_ERROR: // 有消息进来 std::cout << " Message in-> topic:" << msg->topic_name() << ", partition:[" << msg->partition() << "] at offset " << msg->offset() << " key: " << msg->key() << " payload: " << (char *)msg->payload() << std::endl; break; default: std::cerr << "Consumer error: " << msg->errstr() << std::endl; break; } }

4. 位移提交

• consumer需要向kafka汇报自己的位移数据，这个汇报过程被称为位移提交。因为consumer能够同时消费多个分区的数据，所以位移的提交实际上是在分区粒度上进行的，即Consumer需要为分配给它的每个分区提交各自的位移数据。
• 提交位移主要是为了表征consumer的消费进度，这样当consumer发生故障重启之后，就能够从kafka中读取之前提交的位移值。
• 自动提交
  1. enable.auto.commit - 如果是true，将自动定时提交消费者offset，默认是true
  2. auto.commit.interval.ms - 自动提交offset之间的间隔毫秒数，默认为5000
  3. 会有重复消费的问题，因为假如在自动提交间隔期间掉电，再重启，kafka仅会从上次提交的位置进行继续消费，中间的一部分会重复
• 手动提交
  • 手动提交enable.auto.commit一定要设置为false
  1. 同步提交

     1 2 3 4 5 6

     while (true) { RdKafka::Message *msg = m_consumer->consume(1000); // 1s超时 msg_consume(msg, NULL); m_consumer->commitSync(); // 这里会阻塞住，需要等到结果返回才能继续, 影响吞吐量 delete msg; }

  2. 异步提交

     1 2 3 4 5 6

     while (true) { RdKafka::Message *msg = m_consumer->consume(1000); // 1s超时 msg_consume(msg, NULL); m_consumer->commitAsync(); // 不影响吞吐量，但是不知道会不会成功 delete msg; }

  3. 混合使用（常是用这种）

     1 2 3 4 5 6 7 8

     // 消费消息 while (true) { RdKafka::Message *msg = m_consumer->consume(1000); // 1s超时 msg_consume(msg, NULL); m_consumer->commitAsync(); delete msg; } m_consumer->commitSync();

5. 消费者rebalance机制

• 当kafka遇到如下四种情况的时候，kafka会触发Rebalance机制：
  1. 消费组成员发生了变更，比如有新的消费者加入了消费组组或者有消费者宕机
  2. 消费者无法在指定的时间之内完成消息的消费
  3. 消费组订阅的Topic发生了变化
  4. 订阅的Topic的partition发生了变化
• 这里给出一些推荐数值，可以“无脑”地应用在你的生产环境中。
  1. 设置 session.timeout.ms = 6s（表示 consumer 向 broker 发送心跳的超时时间）
  2. 设置 heartbeat.interval.ms = 2s（表示 consumer 每次向 broker 发送心跳的时间间隔）
  3. 要保证 Consumer 实例在被判定为“dead”之前，能够发送至少 3 轮的心跳请求，即session.timeout.ms >= 3 * heartbeat.interval.ms

存储机制和可靠性

存储机制

• kafka固化的数据形式

  1 2 3 4 5

  - test-0（文件夹命名格式topic-partition） - 00000000000012768089.index（索引文件） - 00000000000012768089.log（数据文件） - test-1 -。。。

• 从这个格式也可以看出来，每个分区的offset是独立的，都是从0开始重新编号的。
• consumer寻址过程
  
  1. 根据offset进行二分查找，找到position
  2. 根据索引和position查找消息
     
  • index和log的文件命名，也是offset的起始这样的格式，log文件是有最大的大小限制的，所以可能有两个0000.log，1111.log，那么offset>1111的条目一定不在第一个文件里。

高效性

1. 消息顺序追加写日志
2. PageCache
3. 零拷贝（从磁盘索引到的数据，无需经过应用层，直接从网卡发送出去）
   

可靠性

producer的可靠性保证

• ACK策略配置

  1 2 3 4 5 6

  # -1(全量同步确认，强可靠性保证) Request.required.acks= -1 # 1(leader 确认收到, 默认) Request.required.acks = 1 # 0(不确认，但是吞吐量大) Request.required.acks = 0

设置为CP系统（重点）

1
2
3

request.required.acks=-1
min.insync.replicas = ${N/2 + 1}
unclean.leader.election.enable = false

• 这样设置后只有在所有follower都复制完成后，才回ACK，这样无论哪个机器down掉，都不会丢失数据，满足强一致性要求。
• 异常情况下，当leader复制到f1和f2时，此时leader宕机，此时f3还没有复制，所以f1和f2都有可能成为新的leader(unclean.leader.election.enable设置为false，未同步的机器不能成为leader)，producer会收到异常，重新发送给leader信息（会有重复信息，暂不考虑）。但是此时正常工作的副本数量为1，不满足min.insync.replicas最小服务副本数的配置，此时将会整个系统停止服务。即牺牲可用性停服
  

设置为AP系统

1
2
3

request.required.acks=1
min.insync.replicas = 1
unclean.leader.election.enable = false

• leader接收到信息就返回ack，无需管是否完成了同步

broker的可靠性保证

• HW：用户可以读取到的，已经同步好的位置
• LEO：leader已经写入的位置，此时可能follower可能还没有同步好
• 所以，假设系统不是强一致性（CP）的，用户返回消费成功，实际上可能只是LEO位置已经写好了，但是HW并不等于LEO，此时实际上是不成功的。
  

consumer的可靠性保证

• 位移提交：强一致性，使用同步提交

总结

1. kafka环境搭建
2. kafka基本框架，生产者 分区策略；消费组的消费策略
3. C++操作kafka重点内容
4. kafka命令的练习，./kafka-console-consumer.sh –help
5. kafka存储逻辑，索引index->log 怎么查找索引 零拷贝
6. 生产者 强一致性问题，可用性的
7. 消费者 重复消费的问题，自动提交 手动提交问题
8. broker HW LEO的关系