kafka（四）——生产者流程分析(c++)

前言

• kafka生产者负责将数据发布到kafka集群的主题；
• kafka生产者消息发送方式有两种：
  • 同步发送
  • 异步+回调发送

流程

流程说明：

• Kafka Producer整体可看作是一个异步处理操作；
• 消息发送过程中涉及两个线程：main线程和sender线程；
• main线程负责将消息发送至一个双端队列，sender线程负责从双端队列取消息并发送至kafka broker；

消息可靠性

producer的acks参数表示生产者生产消息时，写入到副本的严格程度。决定了生产者的性能与可靠性。

• 0：生产者发送过来的数据，不等待broker确认，直接发送下一条数据，性能最高，但可能存在丢数据；

• 1：生产者发送过来的数据，等待Leader副本确认后发送下一条数据，性能中等；
  

• -1（all）：生产者发送过来的数据，等待所有副本将数据同步后发送下一条数据，性能最慢，安全性最高；

消息有序性

消息保序策略：按key分区，可以实现局部有序，但这又可能会导致数据倾斜，可根据实际情况选择。

示例：

// 指定消息key,即倒数第二个参数,当有相同的两条消息先后存储同一个key,消费者可按顺序消费到

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
		);

Main线程与Sender线程

Main线程

流程

• 创建消息

// librdkafka源码 rdkafka_msg.c

/* Create message */
rkm = rd_kafka_msg_new0(rkt, force_partition, msgflags, payload, len,
                        key, keylen, msg_opaque, &err, &errnox, NULL, 0,
                        rd_clock());
if (unlikely(!rkm)) {
    /* errno is already set by msg_new() */
    rd_kafka_set_last_error(err, errnox);
    return -1;
}

• 选择分区

/* Partition the message */
err = rd_kafka_msg_partitioner(rkt, rkm, 1);
if (likely(!err)) {
    rd_kafka_set_last_error(0, 0);
    return 0;
}

• 调用拦截器

/* Interceptor: unroll failing messages by triggering on_ack.. */
rkm->rkm_err = err;
rd_kafka_interceptors_on_acknowledgement(rkt->rkt_rk,
                                         &rkm->rkm_rkmessage);

Sender线程

参数说明

流程

• 达到batch.size大小或满足linger.ms时间发送消息；
• 消息发送至的kafka服务器后，如果kafka没有应答，默认每个broker节点队列最多缓存 5 个请求，与“max.in.flight.requests.per.connection”参数有关；
• 如配置了“retries”、“ retry.backoff.ms”参数，消息发送失败由kafka内部自动重试，无需手动在回调函数中重试；

同步和异步流程

同步流程

流程说明

• 通过produce方法将消息推送至双端队列；
• 通过flush方法等待发送结果，如outq_len()大于0，说明存在未发送成功的消息；

代码示例

int KafkaProducer::PushMessage(const std::string &str, const std::string &key)
{
	int32_t len = (int32_t)str.length();
	void *payload = const_cast<void *>(static_cast<const void *>(str.data()));

	// produce 方法，生产和发送单条消息到 Broker
	// 如果不加时间戳，内部会自动加上当前的时间戳
	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
		);

	if (RdKafka::ERR_NO_ERROR != errorCode) 
	{
		// kafka 队列满，等待 100 ms
		if (RdKafka::ERR__QUEUE_FULL == errorCode) 
		{
			m_producer->poll(100);
		}
        
        return -1;
	}
    
    // 同步等待200ms
    m_producer->flush(200);
    if(m_producer->outq_len() > 0)  // 用于调试
    {
        printf("Existed not send message.size:%d\n", m_producer->outq_len());
        return -1;
    }
    
    return 0;
}

异步流程

流程说明

• 设置生产者投递报告回调
• 设置生产者自定义分区策略回调
• 消息发送

代码示例

• 设置生产者投递回调

// 生产者投递报告回调
class ProducerDeliveryReportCb : public RdKafka::DeliveryReportCb 
{
public:
	void dr_cb(RdKafka::Message& message)
	{	
		if (message.err())   // 出错回调
		{
			// TODO
		} 
		else                 // 正常回调
		{  
			// TODO
		}
	}
};

// 设置生产者投递报告回调
m_dr_cb = new ProducerDeliveryReportCb; // 创建投递报告回调
errCode = m_config->set("dr_cb", m_dr_cb, errorStr);    // 异步方式发送数据
if (RdKafka::Conf::CONF_OK != errCode) 
{
    printf("Conf set(dr_cb) failed, errorStr:%s", errorStr.c_str());
    break;
}

• 设置生产者自定义分区策略回调

// 生产者自定义分区策略回调：partitioner_cb
class HashPartitionerCb : public RdKafka::PartitionerCb 
{
public:
	// @brief 返回 topic 中使用 key 的分区，msg_opaque 置 NULL
	// @return 返回分区，(0, partition_cnt)
	int32_t partitioner_cb(const RdKafka::Topic *topic, const std::string *key,
		int32_t partition_cnt, void *msg_opaque) 
	{
		// 用于自定义分区策略：这里用 hash。例：轮询方式：p_id++ % partition_cnt
		int32_t partition_id = generate_hash(key->c_str(), key->size()) % partition_cnt;
		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;
	}
};

// 设置生产者自定义分区策略回调
m_partitioner_cb = new HashPartitionerCb; // 创建自定义分区投递回调
errCode = m_topicConfig->set("partitioner_cb", m_partitioner_cb, errorStr);
if (RdKafka::Conf::CONF_OK != errCode) 
{
    printf("Conf set(partitioner_cb) failed, errorStr:%s", errorStr.c_str());
    break;
}

• 消息发送

注意：此处produce执行成功不代表消息发送成功，需根据dr_cb消息回调结果判断消息是否发送成功。

int KafkaProducer::PushMessage(const std::string &str, const std::string &key)
{
	int32_t len = (int32_t)str.length();
	void *payload = const_cast<void *>(static_cast<const void *>(str.data()));

	// produce 方法，生产和发送单条消息到 Broker
	// 如果不加时间戳，内部会自动加上当前的时间戳
	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
		);

	// 轮询处理
	m_producer->poll(0);
	if (RdKafka::ERR_NO_ERROR != errorCode) 
	{
		// kafka 队列满，等待 100 ms
		if (RdKafka::ERR__QUEUE_FULL == errorCode) 
		{
			m_producer->poll(100);
		}
        
        return -1;
	}
    
    return 0;
}