//if(rd_kafka_conf_set(conf,"debug","none",errstr,sizeof(errstr))!= RD_KAFKA_CONF_OK) { // Carefully evaluate whether to enable debugging for an online environment.
"%% Failed to create new producer: %s\\n", errstr);
return1;
}
/* Signal handler for clean shutdown */
signal(SIGINT, stop);
fprintf(stderr,
"%% Type some text and hit enter to produce message\\n"
"%% Or just hit enter to only serve delivery reports\\n"
"%% Press Ctrl-C or Ctrl-D to exit\\n");
while(run &&fgets(buf,sizeof(buf),stdin)){
size_t len =strlen(buf);
rd_kafka_resp_err_t err;
if(buf[len-1]=='\\n')/* Remove newline */
buf[--len]='\\0';
if(len ==0){
/* Empty line: only serve delivery reports */
rd_kafka_poll(rk,0/*non-blocking */);
continue;
}
/*
* Send/Produce message.
* This is an asynchronous call, on success it will only
* enqueue the message on the internal producer queue.
* The actual delivery attempts to the broker are handled
* by background threads.
* The previously registered delivery report callback
* (dr_msg_cb) is used to signal back to the application
* when the message has been delivered (or failed).
*/
retry:
err =rd_kafka_producev(
/* Producer handle */
rk,
/* Topic name */
RD_KAFKA_V_TOPIC(topic),
/* Make a copy of the payload. */
RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
/* Message value and length */
RD_KAFKA_V_VALUE(buf, len),
/* Per-Message opaque, provided in
* delivery report callback as
* msg_opaque. */
RD_KAFKA_V_OPAQUE(NULL),
/* End sentinel */
RD_KAFKA_V_END);
if(err){
/*
* Failed to *enqueue* message for producing.
*/
fprintf(stderr,
"%% Failed to produce to topic %s: %s\\n",
topic,rd_kafka_err2str(err));
if(err == RD_KAFKA_RESP_ERR__QUEUE_FULL){
/* If the internal queue is full, wait for
* messages to be delivered and then retry.
* The internal queue represents both
* messages to be sent and messages that have
* been sent or failed, awaiting their
* delivery report callback to be called.
*
* The internal queue is limited by the
* configuration property
* queue.buffering.max.messages */
rd_kafka_poll(rk,1000/*block for max 1000ms*/);
goto retry;
}
}else{
fprintf(stderr,"%% Enqueued message (%zd bytes) "
"for topic %s\\n",
len, topic);
}
/* A producer application should continually serve
* the delivery report queue by calling rd_kafka_poll()
* at frequent intervals.
* Either put the poll call in your main loop, or in a
* dedicated thread, or call it after every
* rd_kafka_produce() call.
* Just make sure that rd_kafka_poll() is still called
* during periods where you are not producing any messages
* to make sure previously produced messages have their
* delivery report callback served (and any other callbacks
* you register). */
rd_kafka_poll(rk,0/*non-blocking*/);
}
/* Wait for final messages to be delivered or fail.
* rd_kafka_flush() is an abstraction over rd_kafka_poll() which
* waits for all messages to be delivered. */
fprintf(stderr,"%% Flushing final messages..\\n");
rd_kafka_flush(rk,10*1000/* wait for max 10 seconds */);
/* If the output queue is still not empty there is an issue
* with producing messages to the clusters. */
if(rd_kafka_outq_len(rk)>0)
fprintf(stderr,"%% %d message(s) were not delivered\\n",
rd_kafka_outq_len(rk));
/* Destroy the producer instance */
rd_kafka_destroy(rk);
return0;
}
2. Run the command below to compile producer.c.
gcc -lrdkafka ./producer.c -o producer
3. Run the command below to send messages.
./produce <broker><topic><username><password>
Parameter
Description
broker
Accessed network, which can be copied from the Network column in the Access Mode section in Basic Info on the instance details page in the console.
topic
Topic name, which can be copied in Topic Management on the instance details page in the console.
username
Username in the format of instance ID + # + configured username. The instance ID can be obtained in Basic Info on the instance details page in the CKafka console, and the username is set when the user is created on the User Management tab page in the console.
password
Configured password, which is set when the user is created on the User Management tab page in the console.
The execution result is as follows:
4. On the Topic Management tab page on the instance details page in the CKafka console, select the target topic, and click More > Message Query to view the message just sent.
Step 4. Consume messages
1. Create the consumer.c file.
/*
* librdkafka - Apache Kafka C library
*
* Copyright (c) 2019, Magnus Edenhill
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright notice,
* this list of conditions and the following disclaimer in the documentation
* and/or other materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
Accessed network, which can be copied from the Network column in the Access Mode section in Basic Info on the instance details page in the console.
group.id
Consumer group name. You can customize it. After the demo runs successfully, you can see the consumer on the Consumer Group page.
username
Username in the format of instance ID + # + configured username. The instance ID can be obtained in Basic Info on the instance details page in the CKafka console, and the username is set when the user is created on the User Management tab page in the console.
password
Configured password, which is set when the user is created on the User Management page in the console.
topic1 topic2..
Topic name, which can be copied in Topic Management on the instance details page in the console.
The execution result is as follows:
4. On the Consumer Group tab page on the instance details page in the CKafka console, select the corresponding consumer group name, enter the topic name, and click Query Details to view the consumption details.
Yes
No
Was this page helpful?