kafka C++实现生产者

文章目录

  • 1 Kafka 生产者的逻辑
  • 2 Kafka 的C++ API
    • 2.1 RdKafka::Conf
    • 2.2 RdKafka::Message
    • 2.3 RdKafka::DeliveryReportCb
    • 2.4 RdKafka::Event
    • 2.5 RdKafka::EventCb
    • 2.6 RdKafka::PartitionerCb
    • 2.7 RdKafka::Topic
    • 2.8 RdKafka::Producer（核心）
  • 3 Kafka 生产者客户端开发
    • 3.1 必要的参数配置（bootstrap.servers）
    • 3.2 其他重要的生产者参数
      • 3.2.1 acks
      • 3.2.2 max.request.size
      • 3.2.3 retries 和 retry.backoff.ms
      • 3.2.4 compression.type
      • 3.2.5 connection.max.idle.ms
      • 3.2.6 linger.ms
      • 3.2.7 receive.buffer.bytes
      • 3.2.8 send.buffer.bytes
      • 3.2.9 request.timeout.ms
      • 3.2.10 client.id
      • 3.2.11 batch.size
    • 3.3 创建生产者实例
    • 3.4 消息发送
    • 3.5 完整示例代码
  • 4 总结

1 Kafka 生产者的逻辑

在这里插入图片描述
（1）配置生产者客户端参数。
（2）创建相应的生产者实例。
（3）构建待发送的消息。
（4）发送消息。
（5）关闭生产者实例。

2 Kafka 的C++ API

2.1 RdKafka::Conf

enum ConfType{ CONF_GLOBAL, 	// 全局配置 CONF_TOPIC 		// Topic配置 
};
enum ConfResult{ CONF_UNKNOWN = -2, CONF_INVALID = -1, CONF_OK = 0 
};

static Conf * create(ConfType type);
//创建配置对象。Conf::ConfResult set(const std::string &name, const std::string &value, std::string &errstr);
//设置配置对象的属性值，成功返回CONF_OK，错误时错误信息输出到errstr。Conf::ConfResult set(const std::string &name, DeliveryReportCb *dr_cb, std::string &errstr);
//设置dr_cb属性值。Conf::ConfResult set(const std::string &name, EventCb *event_cb, std::string &errstr);
//设置event_cb属性值。Conf::ConfResult set(const std::string &name, const Conf *topic_conf, std::string &errstr);
//设置用于自动订阅Topic的默认Topic配置。Conf::ConfResult set(const std::string &name, PartitionerCb *partitioner_cb, std::string &errstr);
//设置partitioner_cb属性值，配置对象必须是CONF_TOPIC类型。Conf::ConfResult set(const std::string &name, PartitionerKeyPointerCb *partitioner_kp_cb,std::string &errstr);
//设置partitioner_key_pointer_cb属性值。Conf::ConfResult set(const std::string &name, SocketCb *socket_cb, std::string &errstr);
//设置socket_cb属性值。Conf::ConfResult set(const std::string &name, OpenCb *open_cb, std::string &errstr);
//设置open_cb属性值。Conf::ConfResult set(const std::string &name, RebalanceCb *rebalance_cb, std::string &errstr);
//设置rebalance_cb属性值。Conf::ConfResult set(const std::string &name, OffsetCommitCb *offset_commit_cb, std::string &errstr);
//设置offset_commit_cb属性值。Conf::ConfResult get(const std::string &name, std::string &value) const;
//查询单条属性配置值。

2.2 RdKafka::Message

Message表示一条消费或生产的消息，或是事件。

std::string errstr() const;
//如果消息是一条错误事件，返回错误字符串，否则返回空字符串。ErrorCode err() const;
//如果消息是一条错误事件，返回错误代码，否则返回0。Topic * topic() const;
//返回消息的Topic对象。如果消息的Topic对象没有显示使用RdKafka::Topic::create()创建，需要使用topic_name函数。std::string topic_name() const;
//返回消息的Topic名称。int32_t partition() const;
//如果分区可用，返回分区号。void * payload() const;
//返回消息数据。size_t len() const;
//返回消息数据的长度。const std::string * key() const;
//返回字符串类型的消息key。const void * key_pointer() const;
//返回void类型的消息key。size_t key_len() const;
//返回消息key的二进制长度。int64_t offset () const;
//返回消息或错误的位移。void * msg_opaque() const;
//返回RdKafka::Producer::produce()提供的msg_opaque。virtual MessageTimestamp timestamp() const = 0;
//返回消息时间戳。virtual int64_t latency() const = 0;
//返回produce函数内生产消息的微秒级时间延迟，如果延迟不可用，返回-1。virtual struct rd_kafka_message_s *c_ptr () = 0;
//返回底层数据结构的C rd_kafka_message_t句柄。virtual Status status () const = 0;
//返回消息在Topic Log的持久化状态。virtual RdKafka::Headers *headers () = 0;
//返回消息头。virtual RdKafka::Headers *headers (RdKafka::ErrorCode *err) = 0;
//返回消息头，错误信息会输出到err。

2.3 RdKafka::DeliveryReportCb

每收到一条RdKafka::Producer::produce()函数生产的消息，调用一次投递报告回调函数，RdKafka::Message::err()将会标识Produce请求的结果。
为了使用队列化的投递报告回调函数，必须调用RdKafka::poll()函数。

virtual void dr_cb(Message &message)=0;

当一条消息成功生产或是rdkafka遇到永久失败或是重试次数耗尽，投递报告回调函数会被调用。

C++封装示例：

class ProducerDeliveryReportCb : public RdKafka::DeliveryReportCb
{
public:void dr_cb(RdKafka::Message &message){if(message.err())std::cerr << "Message delivery failed: " << message.errstr() << std::endl;else{// Message delivered to topic test [0] at offset 135000std::cerr << "Message delivered to topic " << message.topic_name()<< " [" << message.partition() << "] at offset "<< message.offset() << std::endl;}}
};

2.4 RdKafka::Event

enum Type{ EVENT_ERROR, //错误条件事件 EVENT_STATS, // Json文档统计事件 EVENT_LOG, // Log消息事件 EVENT_THROTTLE // 来自Broker的throttle级信号事件 
};

virtual Type type() const =0;
//返回事件类型。
virtual ErrorCode err() const =0;
//返回事件错误代码。
virtual Severity severity() const =0;
//返回log严重级别。
virtual std::string fac() const =0;
//返回log基础字符串。
virtual std::string str () const =0;
//返回Log消息字符串。
virtual int throttle_time() const =0;
//返回throttle时间。
virtual std::string broker_name() const =0;
//返回Broker名称。
virtual int broker_id() const =0;
//返回Broker ID。

2.5 RdKafka::EventCb

事件是从RdKafka传递错误、统计信息、日志等消息到应用程序的通用接口。

virtual void event_cb(Event &event)=0; //  事件回调函数

C++封装示例：

class ProducerEventCb : public RdKafka::EventCb
{
public:void event_cb(RdKafka::Event &event){switch(event.type()){case RdKafka::Event::EVENT_ERROR:std::cout << "RdKafka::Event::EVENT_ERROR: " << RdKafka::err2str(event.err()) << std::endl;break;case RdKafka::Event::EVENT_STATS:std::cout << "RdKafka::Event::EVENT_STATS: " << event.str() << std::endl;break;case RdKafka::Event::EVENT_LOG:std::cout << "RdKafka::Event::EVENT_LOG " << event.fac() << std::endl;break;case RdKafka::Event::EVENT_THROTTLE:std::cout << "RdKafka::Event::EVENT_THROTTLE " << event.broker_name() << std::endl;break;}}
};

2.6 RdKafka::PartitionerCb

PartitionerCb用实现自定义分区策略，需要使用RdKafka::Conf::set()设置partitioner_cb属性。

virtual int32_t partitioner_cb(const Topic *topic, const std::string *key, int32_t partition_cnt,void *msg_opaque)=0;
//Partitioner回调函数

返回topic主题中使用key的分区，key可以是NULL或字符串。
partition_cnt表示该主题的分区数量（用于hash计算）
返回值必须在0到partition_cnt间，如果分区失败可能返回RD_KAFKA_PARTITION_UA (-1)。
msg_opaque与RdKafka::Producer::produce()调用提供的msg_opaque相同。

C++封装示例：

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

2.7 RdKafka::Topic

static Topic * create(Handle *base, const std::string &topic_str, Conf *conf, std::string &errstr);
//使用conf配置创建名为topic_str的Topic句柄。const std::string name ();
//获取Topic名称。bool partition_available(int32_t partition) const;
//获取parition分区是否可用，只能在 RdKafka::PartitionerCb回调函数内被调用。ErrorCode offset_store(int32_t partition, int64_t offset);
//存储Topic的partition分区的offset位移，只能用于RdKafka::Consumer，不能用于RdKafka::KafkaConsumer高级接口类。
//使用本接口时，auto.commit.enable参数必须设置为false。virtual struct rd_kafka_topic_s *c_ptr () = 0;
//返回底层数据结构的rd_kafka_topic_t句柄，不推荐利用rd_kafka_topic_t句柄调用C API，但如果C++ API没有提供相应功能，
//可以直接使用C API和librdkafka核心交互。

static const int32_t PARTITION_UA = -1;		//未赋值分区
static const int64_t OFFSET_BEGINNING = -2;	//特殊位移，从开始消费
static const int64_t OFFSET_END = -1;		//特殊位移，从末尾消费
static const int64_t OFFSET_STORED = -1000;	//使用offset存储

2.8 RdKafka::Producer（核心）

static Producer * create(Conf *conf, std::string &errstr);
//创建一个新的Producer客户端对象，conf用于替换默认配置对象，本函数调用后conf可以重用。成功返回新的Producer客户端对象
//，失败返回NULL，errstr可读错误信息。ErrorCode produce(Topic *topic, int32_t partition, int msgflags, void *payload, size_t len,
const std::string *key, void *msg_opaque);
//生产和发送单条消息到Broker。msgflags：可选项为RK_MSG_BLOCK、RK_MSG_FREE、RK_MSG_COPY。

在这里插入图片描述
返回错误码：
在这里插入图片描述

ErrorCode produce(Topic *topic, int32_t partition, int msgflags, void *payload, size_t len,const void *key, 
size_t key_len, void *msg_opaque);
//生产和发送单条消息到Broker，传递key数据指针和key长度。ErrorCode produce(Topic *topic, int32_t partition, const std::vector< char > *payload, 
const std::vector< char > *key, void *msg_opaque);
//生产和发送单条消息到Broker，传递消息数组和key数组。接受数组类型的key和payload，数组会被复制。//ErrorCode flush (int timeout_ms);
//等待所有未完成的所有Produce请求完成。为了确保所有队列和已经执行的Produce请求在中止前完成，flush操作优先于销毁生产者
//实例完成。本函数会调用Producer::poll()函数，因此会触发回调函数。//ErrorCode purge (int purge_flags);
//清理生产者当前处理的消息。本函数调用时可能会阻塞一定时间，当后台线程队列在清理时。应用程序需要在调用poll或flush函数后
//，执行清理消息的dr_cb回调函数。virtual Error *init_transactions (int timeout_ms) = 0;
//初始化Producer实例的事务。失败返回RdKafka::Error错误对象，成功返回NULL。
//通过调用RdKafka::Error::is_retriable()函数可以检查返回的错误对象是否有权限重试，调用
//RdKafka::Error::is_fatal()检查返回的错误对象是否是严重错误。返回的错误对象必须elete。virtual Error *begin_transaction () = 0;
//启动事务。本函数调用前，init_transactions()函数必须被成功调用。
//成功返回NULL，失败返回错误对象。通过调用RdKafka::Error::is_fatal_error()函数可以检查是否是严重错误，返回的错误对象
//必须delete。virtual Error send_offsets_to_transaction (const std::vector &offsets,const ConsumerGroupMetadata*group_metadata,int timeout_ms) = 0;
//发送TopicPartition位移链表到由group_metadata指定的Consumer Group协调器，如果事务提交成功，位移才会被提交。virtual Error *commit_transaction (int timeout_ms) = 0;
//提交当前事务。在实际提交事务时，任何未完成的消息会被完成投递。
//成功返回NULL，失败返回错误对象。通过调用错误对象的方法可以检查是否有权限重试，是否是严重错误、可中止错误等。virtual Error *abort_transaction (int timeout_ms) = 0;
//停止事务。本函数从非严重错误、可终止事务中用于恢复。未完成消息会被清理。

3 Kafka 生产者客户端开发

3.1 必要的参数配置（bootstrap.servers）

（1）指定连接 Kafka 集群所需要的 broker 地址清单，具体的内容格式为 host1:port1,host2:port2，可以设置一个或者多个地址，中间以逗号进行隔开，此参数的默认值为 “”。
（2）注意这里并非需要所有的 broker 地址，因为生产者会从给定的 broker 里查找其他 broker 的信息。
（3）过建议至少要设置两个以上的 broker 地址信息，当其中任意一个宕机时，生产者仍然可以连接到 Kafka 集群上。

// 创建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;
}errCode = m_config->set("statistics.interval.ms", "10000", errorStr);
if(errCode != RdKafka::Conf::CONF_OK)
{std::cout << "Conf set failed:" << errorStr << std::endl;
}errCode = m_config->set("message.max.bytes", "10240000", errorStr);
if(errCode != RdKafka::Conf::CONF_OK)
{std::cout << "Conf set failed:" << errorStr << std::endl;
}
errCode = m_config->set("bootstrap.servers", m_brokers, errorStr);
if(errCode != RdKafka::Conf::CONF_OK)
{std::cout << "Conf set failed:" << errorStr << std::endl;
}

3.2 其他重要的生产者参数

3.2.1 acks

用来指定分区中必须要有多少个副本收到这条消息，之后生产者才会认为这条消 息是成功写入的。 acks
是生产者客户端中一个非常重要的参数 ，它涉及消息的可靠性和吞吐量之间的权衡。acks 参数有3种类
型的值（都是字符串类型）。

  • acks = 1。默认值即为 1。生产者发送消息之后，只要分区的 leader 副本成功写入消息，那么它就
    会收到来自服务端的成功响应。如果消息无法写入 leader 副本，比如在 leader 副本崩溃、重新选
    举新的 leader 副本的过程中，那么生产者就会收到一个错误的响应，为了避免消息丢失，生产者
    可以选择重发消息。如果消息写入 leader 副本并返回成功给生产者，且在被其他 follower 副本拉
    取之前 leader 副本崩溃，那么此时消息还是会丢失，因为新选举的 leader 副本中并没有这条对应
    的消息。
    acks 设置为 1，是消息可靠性和吞吐量之间的折中方案。
  • acks = 0。生产者发送消息之后不需要等待任何服务端的响应。
    如果在消息从发送到写入 Kafka 的过程中出现了某些异常，导致 Kafka 并没有收到这条消息，那么
    生产者也无从得知，消息也就丢失了。
    在其他配置环境相同的情况下，acks 设置为 0 可以达到最大的吞吐量。
  • acks = -1 或 acks = all。生产者在消息发送之后，需要等待 ISR 中的所有副本都成功写入消息之后
    才能够收到来自服务端的成功响应。
    在其他配置环境相同的情况下，acks 设置为 -1 可以达到最强的可靠性。
    但是并不意味着消息就一定可靠，因为 ISR 中可能只有 leader 副本，这样就退化成了 acks = 1 的
    情况。要获得更高的消息可靠性需要配合 min.insync.replicas 等参数的联动。

注意 acks 参数配置的值是一个字符串类型，而不是整数类型。

//范例：
RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
ConfResult ret = conf->set("acks", "1", errstr);
ConfResult ret = conf->set("acks", "0", errstr);
ConfResult ret = conf->set("acks", "all", errstr);

3.2.2 max.request.size

这个参数用来限制生产者客户端能够发送的消息的最大值，默认值为 1048576 B，即 1 MB。
一般情况下，这个默认值就可以满足大多数的应用场景了。
不建议盲目地增大这个参数的配置值，尤其是在对 Kafka 整体脉络没有足够把控的时候。
因为这个参数还涉及一些其他参数的联动，比如 broker 端的 message.max.bytes 参数，如果配
置错误可能会引起一些不必要的一场。
比如讲 broker 端的 message.max.bytes 参数配置为 10， 而 max.request.size 参数配置为 20，
那么当我们发送一条消息大小为 15 的消息时，生产者客户端就会报出异常：
The reqeust included a message larger than the max message size the server will accept.

errCode = conf->set("message.max.bytes", "10240000", errorStr);

3.2.3 retries 和 retry.backoff.ms

retries 重试次数，默认0

retry.backoff.ms 重试间隔，默认100

  • retries 参数用来配置生产者重试的次数，默认值为 0，即发生异常的时候不进行任何的重试动作。
  • 消息在从生产者发出道成功写入服务器之前可能发生一些临时性的异常，比如网络抖动、Leader
    副本的选举等，这种异常往往是可以自行恢复的，生产者可以通过配置 retries 大于 0 的值，以此
    通过内部重试来恢复而不是一味的将异常抛给生产者的应用程序。
  • 如果重试达到设定的次数，那么生产者就会放弃重试并返回异常。
  • 不过并不是所有的异常都是可以通过重试来解决的，比如消息太大，超过 max.request.size 参数 配置的值时，这种方式就不行了。
  • 重试还和另一个参数 retry.backoff.ms 有关，这个参数的默认值为 100，它用来设定两次重试之
    间的时间间隔，避免无效的频繁重试。
  • 在配置 retries 和 retry.backoff.ms 之前，最好先估算一下可能的异常恢复时间，这样可以设定总
    的重试时间大于这个异常恢复时间，以此来避免生产者过早地放弃重试。
  • Kafka 可以保证同一个分区中的消息时有序的。
  • 如果生产者按照一定的顺序发送消息，那么这些消息也会顺序的写入分区，进而消费者也可以按照 同样的顺序消费它们。
  • 对于某些应用来说，顺序性非常重要，比如 Mysql 的 binlog 传输，如果出现错误就会造成非常严 重的后果。
  • 如果讲 retries 参数设置为非零值，并且 max.in.flight.requests.per.connection 参数配置为大于
    1 的值，那么就会出现错序的现象：如果第一批次消息写入失败，而第二批次消息写入成功，那么生
    产者会重试发送第一批次的消息，此时如果第一批次的消息写入成功，那么这两个批次的消息就出 现了错序。
  • 一般而言，在需要保证顺序的场合建议把参数 max.in.flight.requests.per.connection 配置为 1，而不是把
    retries 配置为 0. 不过这样也会影响整体的吞吐。

max.in.flight.requests.per.connection = 1 限制客户端在单个连接上能够发送的未响应请求的个数。设
置此值是1表示kafka broker在响应请求之前client不能再向同一个broker发送请求。注意：设置此参数
是为了避免消息乱序

3.2.4 compression.type

  • 这个参数用来指定消费的压缩方式，默认值为 “none”，即默认情况下，消息不会被压缩。
  • 该参数还可以配置为 “gzip”，“snappy”，“lz4”。
  • 对消息进行压缩可以极大地减少网络传输量、降低网络 I/O ，从而提高整体的性能。
  • 消息压缩是一种使用时间换空间的优化方式，如果对时延有一定的要求，则不推荐对消息进行压
    缩。

3.2.5 connection.max.idle.ms

这个参数用来指定在多久之后关闭闲置的连接，默认值时 540000 ms，即 9 分钟。

3.2.6 linger.ms

  • 这个参数用来指定生产者发送 Producer Batch 之前等待更多消息（ProducerRecord）加入
    ProducerBatch 的时间，默认值为 0。

  • 生产者客户端会在 ProducerBatch 被填满或等待时间超过 linger.ms 值时发送出去。

  • 增大这个参数的值会增加消息的延迟，但是同时能提升一定的吞吐量。

  • 这个 linger.ms 参数与 TCP 协议中的 Nagle 算法有异曲同工之妙。

3.2.7 receive.buffer.bytes

  • 这个参数用来设置 Socket 接受消息缓冲区（SO_RECBUF）的大小，默认值为 32768（B），即 32
    KB。

  • 如果设置为 -1，则使用操作系统的默认值。

  • 如果 Producer 与 Kafka 处于不同的机房，则可以适当调大这个参数值。

3.2.8 send.buffer.bytes

  • 这个参数用来设置 Socket 发送消息缓冲区（SO_SNDBUF）的大小，默认值为 131072 （B），即
    128 KB。
  • 与 receive.buffer.bytes 参数一样，如果设置为 -1 ，则使用操作系统默认值。

3.2.9 request.timeout.ms

  • 这个参数用来配置 Producer 等待请求响应的最长时间，默认值为 30000 ms。
  • 请求超时之后可以选择进行重试。
  • 注意这个参数需要比 broker 端参数 replica.lag.time.max.ms 的值要大，这样可以减少因客户端重
    试而引起的消息重复的概率。
  • 根据具体场景和需求，需要根据网络状况、Kafka集群负载和消息处理要求来调整该参数值。较低延迟要求的场景可以选择较小的值，而对于网络不稳定或处理压力较大的情况，可能需要适当增加该参数值。

3.2.10 client.id

用来设定 KafkaProducer 对应的客户端 id。默认值为 “”。

3.2.11 batch.size

batch.size 是 producer 最重要的参数之一 ！它对于调优 producer 吞吐量和延时性能指标都有着非常
重要的作用 。
producer 会将发往同一分区的多条消息封装进一个 batch中，当 batch 满了的时候， producer 会发送
batch 中的所有消息 。不过， producer并不总是等待batch满了才发送消息，很有可能当batch还有很
多空闲空间时 producer 就发送该 batch 。显然，batch 的大小就显得非常重要 。
通常来说，一个小的 batch 中包含的消息数很少，因而一次发送请求能够写入的消息数也很少，所以
producer 的吞吐量会很低；一个 batch 非常之巨大，那么会给内存使用带来极大的压力，因为不管是
否能够填满，producer 都会为该batch 分配固定大小的内存。
因此batch.size 参数的设置其实是一种时间与空间权衡的体现 。batch.size 参数默认值是 16384 ，即
16KB 。这其实是一个非常保守的数字。 在实际使用过程中合理地增加该参数值，通常都会发现
producer 的吞吐量得到了相应的增加 。

3.3 创建生产者实例

生产者的相关配置和实例的创建可以在类的构造函数实现。比如Kafka Conf对象、Topic Conf对象、设置Broker属性、Producer、Topic对象等。

// 创建Producer
m_producer = RdKafka::Producer::create(m_config, errorStr);
if(m_producer == NULL)
{std::cout << "Create Producer failed:" << errorStr << std::endl;
}
// 创建Topic对象
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.4 消息发送

librdkafka提供的异步的生产接口，异步的消费接口和同步的消息接口，没有同步的生产接口。
同一个生产者可以发送多个主题的，在内部处理时根据传入的topic对象发送给对应的主题分区。

 RdKafka::ErrorCode errorCode = m_producer->produce(m_topic, RdKafka::Topic::PARTITION_UA,RdKafka::Producer::RK_MSG_COPY,payload, len, &key, NULL);

3.5 完整示例代码

KafkaProducer.h

#ifndef KAFKAPRODUCER_H
#define KAFKAPRODUCER_H#pragma once
#include <string>
#include <iostream>
#include "rdkafkacpp.h"class KafkaProducer
{
public:/*** @brief KafkaProducer* @param brokers* @param topic* @param partition*/explicit KafkaProducer(const std::string& brokers, const std::string& topic, int partition);/*** @brief push Message to Kafka* @param str, message data*/void pushMessage(const std::string& str, const std::string& key);~KafkaProducer();private:std::string m_brokers;			// Broker列表，多个使用逗号分隔std::string m_topicStr;			// Topic名称int m_partition;				// 分区RdKafka::Conf* m_config;        // Kafka Conf对象RdKafka::Conf* m_topicConfig;   // Topic Conf对象RdKafka::Topic* m_topic;		// Topic对象RdKafka::Producer* m_producer;	// Producer对象/*只要看到Cb 结尾的类，要继承它然后实现对应的回调函数*/RdKafka::DeliveryReportCb* m_dr_cb;RdKafka::EventCb* m_event_cb;RdKafka::PartitionerCb* m_partitioner_cb;
};#endif

KafkaProducer.cpp

#include "KafkaProducer.h"// call back
class ProducerDeliveryReportCb : public RdKafka::DeliveryReportCb
{
public:void dr_cb(RdKafka::Message &message){if(message.err())std::cerr << "Message delivery failed: " << message.errstr() << std::endl;else{// Message delivered to topic test [0] at offset 135000std::cerr << "Message delivered to topic " << message.topic_name()<< " [" << message.partition() << "] at offset "<< message.offset() << std::endl;}}
};class ProducerEventCb : public RdKafka::EventCb
{
public:void event_cb(RdKafka::Event &event){switch (event.type()){case RdKafka::Event::EVENT_ERROR:std::cout << "RdKafka::Event::EVENT_ERROR: " << RdKafka::err2str(event.err()) << std::endl;break;case RdKafka::Event::EVENT_STATS:std::cout << "RdKafka::Event::EVENT_STATS: " << event.str() << std::endl;break;case RdKafka::Event::EVENT_LOG:std::cout << "RdKafka::Event::EVENT_LOG " << event.fac() << std::endl;break;case RdKafka::Event::EVENT_THROTTLE:std::cout << "RdKafka::Event::EVENT_THROTTLE " << event.broker_name() << std::endl;break;}}
};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;}
};KafkaProducer::KafkaProducer(const std::string& brokers, const std::string& topic, int partition)
{m_brokers = brokers;m_topicStr = topic;m_partition = partition;/* 创建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;std::string errorStr;m_dr_cb = new ProducerDeliveryReportCb;// 设置dr_cb属性值errCode = m_config->set("dr_cb", m_dr_cb, errorStr);if (errCode != RdKafka::Conf::CONF_OK){std::cout << "Conf set failed:" << errorStr << std::endl;}// 设置event_cb属性值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;}// 设置配置对象的属性值errCode = m_config->set("statistics.interval.ms", "10000", errorStr);if (errCode != RdKafka::Conf::CONF_OK){std::cout << "Conf set failed:" << errorStr << std::endl;}errCode = m_config->set("message.max.bytes", "10240000", errorStr);if (errCode != RdKafka::Conf::CONF_OK){std::cout << "Conf set failed:" << errorStr << std::endl;}errCode = m_config->set("bootstrap.servers", m_brokers, errorStr);if (errCode != RdKafka::Conf::CONF_OK){std::cout << "Conf set failed:" << errorStr << std::endl;}/* 创建Producer */m_producer = RdKafka::Producer::create(m_config, errorStr);if (m_producer == NULL){std::cout << "Create Producer failed:" << errorStr << std::endl;}/* 创建Topic对象 */m_topic = RdKafka::Topic::create(m_producer, m_topicStr, m_topicConfig, errorStr);if (m_topic == NULL){std::cout << "Create Topic failed:" << errorStr << std::endl;}
}KafkaProducer::~KafkaProducer()
{while (m_producer->outq_len() > 0){std::cerr << "Waiting for " << m_producer->outq_len() << std::endl;m_producer->flush(5000);}delete m_config;delete m_topicConfig;delete m_topic;delete m_producer;delete m_dr_cb;delete m_event_cb;delete m_partitioner_cb;
}void KafkaProducer::pushMessage(const std::string& str, const std::string& key)
{int32_t len = str.length();void* payload = const_cast<void*>(static_cast<const void*>(str.data()));RdKafka::ErrorCode errorCode = m_producer->produce(m_topic,RdKafka::Topic::PARTITION_UA,RdKafka::Producer::RK_MSG_COPY,payload,len,&key,NULL);m_producer->poll(0);if (errorCode != RdKafka::ERR_NO_ERROR){std::cerr << "Produce failed: " << RdKafka::err2str(errorCode) << std::endl;if (errorCode == RdKafka::ERR__QUEUE_FULL){m_producer->poll(100);}}
}

CMakeLists.txt

cmake_minimum_required(VERSION 2.8)project(KafkaProducer)set(CMAKE_CXX_STANDARD 11)
set(CMAKE_CXX_COMPILER "g++")
set(CMAKE_CXX_FLAGS "-std=c++11 ${CMAKE_CXX_FLAGS}")
set(CMAKE_INCLUDE_CURRENT_DIR ON)# Kafka头文件路径
include_directories(/usr/local/include/librdkafka)
# Kafka库路径
link_directories(/usr/local/lib)aux_source_directory(. SOURCE)add_executable(${PROJECT_NAME} ${SOURCE})
TARGET_LINK_LIBRARIES(${PROJECT_NAME} rdkafka++)

测试文件main.cpp

#include <iostream>
#include "KafkaProducer.h"
using namespace std;int main()
{// 创建Producer// KafkaProducer producer("127.0.0.1:9092,192.168.2.111:9092", "test", 0);KafkaProducer producer("127.0.0.1:9092", "test", 0);for(int i = 0; i < 10000; i++){char msg[64] = {0};sprintf(msg, "%s%4d", "Hello RdKafka ", i);// 生产消息char key[8] = {0};      // 主要用来做负载均衡sprintf(key, "%d", i);producer.pushMessage(msg, key);  }RdKafka::wait_destroyed(5000);
}

编译：

mkdir build
cd build
cmake ..
make

4 总结

Kafka Producer使用流程：

  • 创建Kafka配置实例。
  • 创建Topic配置实例。
  • 设置Kafka配置实例Broker属性。
  • 设置Topic配置实例属性。
  • 注册回调函数（分区策略回调函数需要注册到Topic配置实例）。
  • 创建Kafka Producer客户端实例。
  • 创建Topic实例。
  • 阻塞等待Producer生产消息完成。
  • 等待Produce请求完成。
  • 销毁Kafka Producer客户端实例。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/182553.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【代码】考虑灵活性供需平衡的电力系统优化调度模型

【代码】考虑灵活性供需平衡的电力系统优化调度模型

程序名称&#xff1a;考虑灵活性供需平衡的电力系统优化调度模型 实现平台&#xff1a;matlab-yalmip-cplex/gurobi 代码简介&#xff1a;最可再生能源发电设备和并网技术快速发展&#xff0c;以新能源为主导的新型电力系统逐步形成。高比例新能源的随机波动性导致电力系统的…
阅读更多...
redis的数据类型的操作增删改查

redis的数据类型的操作增删改查

redis的数据类型的操作增删改查 redis的高可用&#xff1a; 在集群当中有一个非常重要的指标&#xff0c;提供正常服务的时间的百分比&#xff08;365天&#xff09;99.9% redis的高可用的含义要更加宽泛&#xff0c;正常服务是指标之一数据容量扩展&#xff0c;数据的安全性…
阅读更多...
Oracle 11g安装过程

Oracle 11g安装过程

文章目录 前言1.下载安装包2.安装2.1本地安装文件2.2 安装过程 3.查看是否安装成功3.1 查看oracle是否安装成功3.2 查看oracle服务 前言 本文仅用于记录亲自安装oracle的过程 1.下载安装包 官网地址&#xff1a; Oracle Database 11g Release 2 (11.2.0.1.0) 注意&#xff…
阅读更多...
数据治理框架和成熟度模型

数据治理框架和成熟度模型

数据治理成熟度模型 一个企业的数据治理能力越高&#xff0c;所享受到数据治理带来的价值也会越多&#xff0c;如增加收入、减少成本、降低风险等。于是&#xff0c;很多企业想要准确地评估本公司的数据治理能力&#xff0c;可以利用数据治理成熟度模型方法&#xff0c;包括 D…
阅读更多...
Javaweb之Vue组件库Element案例的详细解析

Javaweb之Vue组件库Element案例的详细解析

4.4 案例 4.4.1 案例需求 参考 资料/页面原型/tlias智能学习辅助系统/首页.html 文件&#xff0c;浏览器打开&#xff0c;点击页面中的左侧栏的员工管理&#xff0c;如下所示&#xff1a; 需求说明&#xff1a; 制作类似格式的页面 即上面是标题&#xff0c;左侧栏是导航&…
阅读更多...
泛微OA对接金蝶云星空方案分享（对接场景解析）

泛微OA对接金蝶云星空方案分享（对接场景解析）

分享金蝶云星空跟泛微OA系统集成对接的方案分享&#xff0c;主讲审批流程对接&#xff0c;表单对接的两类场景。分别是金蝶云星空发起申请和泛微发起流程审批&#xff0c;最终实现统一管理。 数据集成主要有以下好处&#xff1a; &#xff08;1&#xff09;数据一致性&#xf…
阅读更多...
吉利展厅 | 透明OLED拼接2x2：科技与艺术的完美融合

吉利展厅 | 透明OLED拼接2x2：科技与艺术的完美融合

产品&#xff1a;4块55寸OLED透明拼接屏 项目地点&#xff1a;南宁 项目时间&#xff1a;2023年11月 应用场景&#xff1a;吉利展厅 在2023年11月的南宁&#xff0c;吉利展厅以其独特的展示设计吸引了众多参观者的目光。其中最引人注目的亮点是展厅中央一个由四块55寸OLED透…
阅读更多...
函数的极值与最值

函数的极值与最值

函数的最值 1.闭区间上连续函数的最值 1.求驻点或不可导点&#xff08;可能的极值点&#xff09; 2.求函数在驻点&#xff0c;不可导点&#xff0c;端点的函数值 3.比较大小 例题&#xff1a; 例题思想&#xff1a;分段函数分段点必须验证导数的存在性 几种常见的最值类型 1.…
阅读更多...
WordPress自动采集伪原创发布工具

WordPress自动采集伪原创发布工具

在当今数字化时代&#xff0c;随着信息爆炸式增长&#xff0c;网站内容的更新速度飞快。对于拥有WordPress网站的用户而言&#xff0c;如何轻松而又快速地批量采集伪原创内容成为一项具有挑战性的任务。本文将专心分享一些方法和技巧&#xff0c;帮助WordPress用户实现批量采集…
阅读更多...
软件设计模式原则（三）单一职责原则

软件设计模式原则（三）单一职责原则

单一职责原则&#xff08;SRP&#xff09;又称单一功能原则。它规定一个类应该只有一个发生变化的原因。所谓职责是指类变化的原因。如果一个类有多于一个的动机被改变&#xff0c;那么这个类就具有多于一个的职责。而单一职责原则就是指一个类或者模块应该有且只有一个改变的原…
阅读更多...
腿姐技巧（毛中特）

腿姐技巧（毛中特）

肌肉记忆的背
阅读更多...
王道p150 14.假设二叉树采用二叉链表存储结构，设计一个算法，求非空二叉树 b的宽度(即具有结点数最多的那一层的结点个数) （c语言代码实现）

王道p150 14.假设二叉树采用二叉链表存储结构，设计一个算法，求非空二叉树 b的宽度(即具有结点数最多的那一层的结点个数) （c语言代码实现）

采用层次遍历的方法求出所有结点的层次&#xff0c;并将所有结点和对应的层次放在一个队列中。然后通过扫描队列求出各层的结点总数&#xff0c;最大的层结点总数即为二叉树的宽度。 /* A B C D E F …
阅读更多...
生成式AI与预测式AI的主要区别与实际应用

生成式AI与预测式AI的主要区别与实际应用

近年来&#xff0c;预测式人工智能&#xff08;Predictive AI&#xff09;通过先进的推荐算法、风险评估模型、以及欺诈检测工具&#xff0c;一直在推高着该领域公司的投资回报率。然而&#xff0c;今年初突然杀出的生成式人工智能&#xff08;Generative AI&#xff09;突然成…
阅读更多...
VMware如何导出和导入镜像OVF虚拟机（以unbuntu为例）

VMware如何导出和导入镜像OVF虚拟机（以unbuntu为例）

前言&#xff1a;我下载了一个unbuntu的系统镜像&#xff0c;基于这个镜像创建了一个unbuntu的虚拟机&#xff0c;辛辛苦苦配置需要的开发环境&#xff0c;如&#xff1a;安装了mysql&#xff0c;安装了jdk等&#xff0c;此时&#xff0c;我的同事可能需要也是同一套类似环境&a…
阅读更多...
MySQL 中的锁（二）

MySQL 中的锁（二）

8.4. 意向锁 但是在上面的例子这里头有两个问题&#xff1a; 如果我们想对大楼整体上 S 锁&#xff0c;首先需要确保大楼中的没有正在维修的楼层&#xff0c;如果有正在维修的楼层&#xff0c;需要等到维修结束才可以对大楼整体上 S 锁。 如果我们想对大楼整体上 X 锁&#xf…
阅读更多...
现代C++之万能引用、完美转发、引用折叠

现代C++之万能引用、完美转发、引用折叠

现代C之万能引用、完美转发、引用折叠 0.导语1.问题引入2.引入万能引用3.万能引用出现场合4.理解左值与右值4.1 精简版4.2 完整版4.3 生命周期延长4.4 生命周期延长应用5.区分万能引用6.表达式的左右值性与类型无关7.引用折叠和完美转发7.1 引用折叠之本质细节7.2 示例与使用7.…
阅读更多...
linux 讨论题合集（个人复习）

linux 讨论题合集（个人复习）

常规文件的权限是什么&#xff1f;如何分配或修改这些权限&#xff1f;文件夹&#xff08;目录&#xff09;的权限是什么&#xff1f;显示常规文件和文件夹的区别 讨论&#xff1a;①常规的文件权限有四种&#xff0c;r可读、w可写、x可执行、-没有权限&#xff1b;②可以使用c…
阅读更多...
鸿蒙4.0开发笔记之ArkTS装饰器语法基础@Extend扩展组件样式与stateStyles多态样式（十一）

鸿蒙4.0开发笔记之ArkTS装饰器语法基础@Extend扩展组件样式与stateStyles多态样式（十一）

一、Extend扩展组件样式 1、作用 前文提到可以使用Styles用于样式的扩展&#xff0c;在Styles的基础上&#xff0c;ArkTS语法还提供了Extend&#xff0c;⽤于扩展原生组件样式&#xff0c;包括Text、Button等等。 2、定义语法 Extend(UIComponentName) function functionNam…
阅读更多...
错误 LNK2001 无法解析的外部符号 __imp__CrtDbgReport

错误 LNK2001 无法解析的外部符号 __imp__CrtDbgReport

”属性“ -->”设置“ --> ”c“ – > ”代码生成“ --> ”运行库“ &#xff0c;将 ”多线程(MT)“ 改为 ”多线程(MTD)“。
阅读更多...
C语言——从终端（键盘）读入 20 个数据到数组中，统计其中正数的个数，并计算这些正数之和

C语言——从终端（键盘）读入 20 个数据到数组中，统计其中正数的个数，并计算这些正数之和

#define _CRT_SECURE_NO_WARNINGS 1#include<stdio.h> int main() {int i0;int sum0;int count0;int arr[20];printf("输入20个数据&#xff1a;\n");for(i1;i<20;i){scanf("%d",&arr[i]);if(arr[i]>0){count;sumarr[i];}}printf("正…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023