当前位置：首页 > SEO教程 >

郑州网站推广费用是多少？了解网站开发相关知识的渠道有哪些？

96SEO 2026-02-20 10:45 11

首先要编写生产者的代码得先知道生产者的逻辑在代码上是怎么体现的

怎么理解呢

配置完参数之后就是创建生产者实例那么实例化生产者之后就是准备生产者

生产消息那么我们在生产者生产消息的时候肯定要初始化和构建消息对象发过去

因为用对象的方式去管理消息更容易拓展和后期进行维护和管理以及消费者读取消息也

不容易出错那么构建完消息对象之后那么就需要将消息对象交给生产者让生产者

生产到指定的kafka的topic中的消息队列也就是topic中的partition分区中因为每个

分区都是独立的队列,生产到消息队列就是发送消息到了消息队列就等消费者进行消费了

配置参数

在实例化生产者对象之前你需要配置生产者的参数。

这一般通过创建一个

RdKafka::Conf

服务器的地址、消息传递语义例如至少一次交付、精确一次交付等、序列化器、分区器等。

这个

Conf

RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

设置其他配置项...RdKafka::Producer::create(conf);实例化生产者

Kafka

函数实现的。

传递配置对象作为参数确保生产者在创建时拥有所需的配置。

构建消息对象

在生产者准备好之后你可以构建消息对象。

这通常包括指定消息的主题、键、内容等信息。

RdKafka::Message

RdKafka::Producer

RdKafka::Producer::create(conf);

RdKafka::Message

方法会返回一个错误码你可以通过检查这个错误码来了解消息发送的状态。

为了确保消息的投递报告RdKafka::DeliveryReportCb回调被调用你需要定期调用

RdKafka::poll()。

这个操作通常在一个独立的线程中完成以确保消息报告的及时处理。

下面是基于Message对象的接口(有些内容都封装在message里)

std::string

//如果消息是一条错误事件返回错误字符串否则返回空字符串。

ErrorCode

err()

//返回消息的Topic对象。

如果消息的Topic对象没有显示使用RdKafka::Topic::create()创建需要使用topic_name函数。

std::string

topic_name()

//返回RdKafka::Producer::produce()提供的msg_opaque。

virtual

MessageTimestamp

//返回produce函数内生产消息的微秒级时间延迟如果延迟不可用返回-1。

virtual

struct

通过这个函数可以获取消息的错误码用于检查消息在生产或消费过程中是否发生了错误。

len():

消息时最重要的信息。

通过这些信息你可以检查消息的状态了解消息的来源和内容以及在消费者端追踪消息的位置。

其他的一些属性比如

2.3

每收到一条RdKafka::Producer::produce()函数生产的消息调用一次投递报告回调函数RdKafka::Message::err()将会标识Produce请求的结果。

为了使用队列化的投递报告回调函数必须调用RdKafka::poll()函数。

投递报告函数RdKafka::DeliveryReportCb在

Kafka

生产者发送消息后用于接收有关消息传递结果的回调通知。

它的主要作用是确保消息是否成功投递到

Kafka

服务器投递报告函数被调用。

这允许你知道消息是否已经成功到达服务器。

追踪消息传递状态

投递报告函数提供了有关消息传递状态的信息。

通过检查消息的错误码通过

获取你可以了解消息是否成功投递到分区以及可能的错误原因比如消息发送超时、分区不存在等等。

确保消息处理

这个回调函数可以帮助你确保消息得到了处理无论是成功发送还是出现了一些错误。

通过错误码你可以适当地处理消息发送过程中的问题例如重试、记录错误日志或者执行其他补救措施。

在整个流程中投递报告函数是为了提供消息传递的状态和结果。

它允许你追踪消息的处理情况确保消息被成功地发送到了

Kafka

服务器并且在出现问题时能够及时地得到通知和处理。

因此在实际的生产环境中及时处理这个回调函数非常重要以保证消息的可靠传递

virtual

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

C封装示例

message){if(message.err())std::cerr

Message

是一个用于表示Kafka事件的类它封装了与事件相关的信息。

在你的描述中列举了事件的不同类型如EVENT_ERROR、EVENT_STATS、EVENT_LOG和EVENT_THROTTLE。

每个事件都有相应的属性和方法来获取事件的类型、错误代码、日志信息等。

enum

返回事件的类型。

类型包括错误条件事件EVENT_ERROR、JSON文档统计事件EVENT_STATS、日志消息事件EVENT_LOG以及来自Broker的throttle级信号事件EVENT_THROTTLE。

err():

如果事件类型是throttle级信号事件返回throttle的时间。

broker_name():

一个抽象基类它定义了一个事件回调函数用于处理RdKafka::Event。

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

virtual

event){switch(event.type()){case

RdKafka::Event::EVENT_ERROR:std::cout

std::endl;break;case

RdKafka::Event::EVENT_STATS:std::cout

event.str()

RdKafka::Event::EVENT_LOG:std::cout

event.fac()

RdKafka::Event::EVENT_THROTTLE:std::cout

event.broker_name()

};这个回调函数的作用是在发生Kafka事件时被调用将相应的RdKafka::Event对象传递给应用程序。

应用程序可以实现自己的RdKafka::EventCb子类然后在这个子类中实现event_cb方法以处理具体的事件。

这样当有错误、统计信息、日志或来自Broker的throttle级信号事件发生时那么逻辑就变成如下

配置参数

你首先配置好生产者的参数这包括Kafka集群的地址、topic的配置等。

创建生产者实例

在生产消息之前你可能需要进行一些准备工作比如初始化和构建消息对象。

生产消息

将构建好的消息对象交给生产者让生产者将消息发送到指定的Kafka

处理事件

这就是上述RdKafka::Event和RdKafka::EventCb的作用了。

在生产者的生命周期中可能会发生一些异步事件如错误、日志信息等。

通过设置RdKafka::EventCb你可以在相应的事件发生时得到通知从而执行你自己的处理逻辑。

关闭生产者

RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);//

eventCallback;conf-set(event_cb,

eventCallback,

RdKafka::Producer::create(conf,

errstr);if

partitioner_cb它会在生产消息并决定消息应该发送到Kafka主题的哪个分区时被调用。

当你在生产消息时你可能希望某些特定的逻辑来决定消息应该发送到哪个分区而不是使用默认的分区策略。

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

virtual

//Partitioner回调函数返回topic主题中使用key的分区key可以是NULL或字符串。

partition_cnt表示该主题的分区数量用于hash计算

返回值必须在0到partition_cnt间如果分区失败可能返回RD_KAFKA_PARTITION_UA

msg_opaque与RdKafka::Producer::produce()调用提供的msg_opaque相同。

partition_cnt用于帮助决定消息将会被分发到哪个分区。

一个指向消息不透明数据的指针

msg_opaque

回调函数需要返回一个整数值表示消息应该发送到的分区。

这个返回值必须介于0到

partition_cnt

回调函数的作用是当生产者在发送消息到Kafka主题时需要决定消息发送到哪个分区时会调用这个函数。

你可以根据你自己的逻辑实现这个回调函数让它根据消息的键或其他特征来决定消息应该发送到哪个分区。

这样你就可以自定义消息的分区策略。

RdKafka::Conf::set()

属性指定自定义的分区策略函数。

然后当你使用生产者发送消息时Kafka客户端会调用你定义的

partitioner_cb

一旦消息被分配到相应的分区生产者就会将消息发送到该分区的消息队列中。

消费者可以从这些分区的队列中读取消息。

C封装示例

[topic][key][partition_cnt][partition_id]

:[test][6419][2][1]sprintf(msg,

partition_id:[%d],

在这里你可以根据消息的键key或其他标准来决定消息应该分发到哪个分区//

等参数进行逻辑判断//

RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);//

partitioner_callback;conf-set(partitioner_cb,

errstr);//

RdKafka::Producer::create(conf,

errstr);if

创建消息对象RdKafka::Producer::Message

msg(my_topic,

RdKafka::Producer::RK_MSG_COPY,

payload

//使用conf配置创建名为topic_str的Topic句柄。

const

std::string

RdKafka::PartitionerCb回调函数内被调用。

ErrorCode

offset_store(int32_t

//存储Topic的partition分区的offset位移只能用于RdKafka::Consumer不能用于RdKafka::KafkaConsumer高级接口类。

//使用本接口时auto.commit.enable参数必须设置为false。

virtual

struct

//返回底层数据结构的rd_kafka_topic_t句柄不推荐利用rd_kafka_topic_t句柄调用C

API但如果C

这个常量代表未指定分区。

在某些情况下如果不想将消费者与特定分区绑定可以使用这个常量表示未赋值分区。

OFFSET_BEGINNING

这个常量用于表示从分区的末尾最新消息开始消费。

如果希望消费者从主题中最新的消息开始消费可以使用此常量。

OFFSET_STORED

这个常量表示使用存储的偏移量进行消费。

有时候消费者可能会将消费的偏移量存储在某个地方比如外部存储、数据库等以便稍后继续从这个位置开始消费。

这个常量可以帮助消费者指定使用存储的偏移量作为消费的起始位置。

Kafka

主题消息时可以根据需要选择不同的起始位置或分区以满足特定的业务需求。

伪代码示例

RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);//

设置消费者配置参数...//

RdKafka::Consumer::create(conf,

errstr);if

RdKafka::Topic::create(consumer,

your_topic,

RdKafka::Topic::OFFSET_BEGINNING;

从起始位置开始消费//

consumer-assign({RdKafka::TopicPartition(your_topic,

partition,

未指定分区讲得是表示消费者并没有指定要消费的具体分区因此消费者将会被动态地分配到可用的分区中。

实际上这种方式可以让消费者根据负载均衡策略被均匀地分配到不同的分区以提高整体的消费效率。

如果是上述代码的话消费者去消费的话不会去特定分区去读取数据而是根据kafka的消费者的分配策略其实分配策略就是负载均衡策略机制被分配到消费者订阅的topic

1.创建

在消息生产之前你描述了初始化和构建消息对象的过程。

这可能涉及到创建一个消息对象设置消息的内容、键、分区等属性。

这样的消息对象可以使用

Kafka

//创建一个新的Producer客户端对象conf用于替换默认配置对象本函数调用后conf可以重用。

成功返回新的Producer客户端对象

//失败返回NULLerrstr可读错误信息。

ErrorCode

produce(Topic

//生产和发送单条消息到Broker。

msgflags可选项为RK_MSG_BLOCK、RK_MSG_FREE、RK_MSG_COPY。

返回错误码

ErrorCode

//生产和发送单条消息到Broker传递key数据指针和key长度。

ErrorCode

produce(Topic

//生产和发送单条消息到Broker传递消息数组和key数组。

接受数组类型的key和payload数组会被复制。

//ErrorCode

flush

//等待所有未完成的所有Produce请求完成。

为了确保所有队列和已经执行的Produce请求在中止前完成flush操作优先于销毁生产者

//实例完成。

本函数会调用Producer::poll()函数因此会触发回调函数。

//ErrorCode

purge

//清理生产者当前处理的消息。

本函数调用时可能会阻塞一定时间当后台线程队列在清理时。

应用程序需要在调用poll或flush函数后

Error

//初始化Producer实例的事务。

失败返回RdKafka::Error错误对象成功返回NULL。

//通过调用RdKafka::Error::is_retriable()函数可以检查返回的错误对象是否有权限重试调用

//RdKafka::Error::is_fatal()检查返回的错误对象是否是严重错误。

返回的错误对象必须elete。

virtual

Error

//启动事务。

本函数调用前init_transactions()函数必须被成功调用。

//成功返回NULL失败返回错误对象。

通过调用RdKafka::Error::is_fatal_error()函数可以检查是否是严重错误返回的错误对象

Error

ConsumerGroupMetadata*group_metadata,int

timeout_ms)

//发送TopicPartition位移链表到由group_metadata指定的Consumer

Group协调器如果事务提交成功位移才会被提交。

virtual

Error

//提交当前事务。

在实际提交事务时任何未完成的消息会被完成投递。

//成功返回NULL失败返回错误对象。

通过调用错误对象的方法可以检查是否有权限重试是否是严重错误、可中止错误等。

virtual

Error

//停止事务。

本函数从非严重错误、可终止事务中用于恢复。

未完成消息会被清理。

Kafka

host1:port1,host2:port2可以设置一个或者多个地址中间以逗号进行隔开此参数的默认值为

broker

就是我们生产者producer在连接kafka集群当中的话可以连接kafka集群当中的一个kafka服务器

如下图示例

RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

if(m_config

RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

if(m_topicConfig

m_topicConfig-set(partitioner_cb,

m_partitioner_cb,

m_config-set(statistics.interval.ms,

10000,

m_config-set(message.max.bytes,

10240000,

m_config-set(bootstrap.servers,

m_brokers,

用来指定分区中必须要有多少个副本收到这条消息之后生产者才会认为这条消

1.acks

副本的过程中那么生产者就会收到一个错误的响应为了避免消息丢失生产者

leader

中的所有副本都成功写入消息之后才能够收到来自服务端的成功响应。

设置为

RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

ConfResult

消息在从生产者发出道成功写入服务器之前可能发生一些临时性的异常比如网络抖动、Leader副本的选举等这种异常往往是可以自行恢复的生产者可以通过配置

retries

的值以此通过内部重试来恢复而不是一味的将异常抛给生产者的应用程序。

临时性异常处理

之前最好先估算一下可能的异常恢复时间这样可以设定总的重试时间大于这个异常恢复时间以此来避免生产者过早地放弃重试。