Message Queue for Kafka
Message Queue for Kafka
Document
- Documentation
Message Queue for KafkaSDK ReferenceApache Kafka SDKJava client SDKPLAINTEXT endpoint sending and receiving messages
Copy page
Download PDFJava client SDK
PLAINTEXT endpoint sending and receiving messages
Copy pageDownload PDFPLAINTEXT endpoint sending and receiving messages
Based on Java client, this topic introduces how to connect to Message Queue for Kafka via the PLAINTEXT endpoint in a VPC environment and how to produce and consume messages.
Prerequisites
1 Install Java dependencies
Add the related dependencies to the Java project's pom.xml. Here is an example of Kafka version 2.2.2.
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.2.2</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.6</version>
</dependency>
2 Add configuration file
- Create the Message Queue for Kafka configuration file config.properties.
- For detailed descriptions of the configuration file fields, refer to configuration file.
- Below is an example configuration file when using the PLAINTEXT endpoint.
- bootstrap.servers=xxxxxsecurity.protocol=PLAINTEXTtopic=my-topicconsumer.group.id=testconsumer.auto.offset.reset=earliestconsumer.enable.auto.commit=falseclient.dns.lookup=use_all_dns_ips
- Create the configuration file loader KafkaConfigurer.java.
- package com.volcengine.openservice.kafka;import java.io.FileInputStream;import java.io.InputStream;import java.util.Properties;public class KafkaConfigurer {private static Properties properties;public synchronized static Properties getKafkaProperties(String path) {if (null != properties) {return properties;}// Obtain the content of the configuration file config.properties.Properties kafkaProperties = new Properties();try {FileInputStream conf = new FileInputStream(path);kafkaProperties.load(conf);} catch (Exception e) {// If loading file fails, consider exiting the program.e.printStackTrace();}properties = kafkaProperties;return kafkaProperties;}}
3 Produce messages
Implementation method
- Create message production program ProducerDemo.java.
- Compile and run ProducerDemo.java to produce messages.
- check the running results.
- the running result example is as follows.
- Produce ok:sasl-0@0Produce ok:sasl-0@1Produce ok:sasl-0@2Produce ok:sasl-0@3
tip
Message Queue for Kafka provides an example Demo for quick access. After downloading and extracting the demo, you can directly execute the following commands to produce messages.
java -cp kafka-demo.jar com.volcengine.openservice.kafka.ProducerDemo ./config.properties
Example code
Example code for producing messages via PLAINTEXT endpoint is as follows. You can also refer to the sample file in the Demo at {DemoPath}/src/main/java/com/volcengine/openservice/kafka/ProducerDemo.java to implement related business logic.
package com.volcengine.openservice.kafka;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
class Producer {
// topic used by the producer
private static String topic;
// configuration used by the producer
private static Properties props = new Properties();
private static KafkaProducer<String, String> producer;
// construct the producer
public Producer(Properties kafkaProperties) {
setProps(kafkaProperties);
setSasl(kafkaProperties);
setTopic(kafkaProperties);
newProducer();
}
// Set the topic that needs to be produced and fill it in here after you apply for it in the console.
private void setTopic(Properties kafkaProperties) {
topic = kafkaProperties.getProperty("topic");
}
// Set the security protocol. Kafka supports SASL_PLAINTEXT and PLAINTEXT protocols
private void setSasl(Properties kafkaProperties) {
String protocol = kafkaProperties.getProperty("security.protocol");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, protocol);
// If the security protocol is PLAINTEXT, no username or password is needed; return directly.
if (protocol.equals("PLAINTEXT")) {
return;
}
// If the security protocol is SASL_PLAINTEXT, obtain the encryption type and configuration of sasl
if (protocol.equals("SASL_PLAINTEXT")) {
props.put(SaslConfigs.SASL_MECHANISM, kafkaProperties.getProperty("sasl.mechanism"));
props.put(SaslConfigs.SASL_JAAS_CONFIG, kafkaProperties.getProperty("sasl.jaas.config"));
return;
}
throw new IllegalArgumentException("security.protocol is not correct");
}
// set the startup parameters for the producer
private void setProps(Properties kafkaProperties) {
// Set the endpoint. Obtain the endpoint for the corresponding topic from the console.
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
//Serialization method of Kafka message
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//Maximum request wait time
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
//Set the client's internal retry count
props.put(ProducerConfig.RETRIES_CONFIG, 5);
//Set the client's internal retry interval
props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
//Set the DNS policy of client
props.put(ProducerConfig.CLIENT_DNS_LOOKUP_CONFIG, kafkaProperties.getProperty("client.dns.lookup"));
}
//Create a producer object,i.e., a production instance
private void newProducer() {
//Create a Producer object. Note that this object is thread-safe. In general, within one process, one producer object is sufficient.
//To improve performance, you can create multiple objects, but not too many. We recommend that you create no more than five objects.
producer = new KafkaProducer<String, String>(props);
}
public void Produce() {
//Create a Kafka message
String value = "this is demo message "; //Message content
//Fetching futures in bulk can speed up the process. However, try to avoid making the batch too large.
List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);
for (int i =0; i < 100; i++) {
//Send message and obtain a Future object
ProducerRecord<String, String> kafkaMessage = new ProducerRecord<String, String>(topic, value + ": " + i);
Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
futures.add(metadataFuture);
}
producer.flush();
for (Future<RecordMetadata> future: futures) {
//Synchronously get the result of the Future object
try {
RecordMetadata recordMetadata = future.get();
System.out.println("Produce ok:" + recordMetadata.toString());
} catch (Throwable t) {
t.printStackTrace();
}
}
}
}
public class ProducerDemo {
public static void main(String args[]) {
if (args.length < 1) {
System.out.println("please input config path");
return;
}
//Get the producer object
Producer producer = new Producer(KafkaConfigurer.getKafkaProperties(args[0]));
try {
//Produce message
producer.Produce();
} catch (Exception e) {
//If message production fails after the client internal retry, the business needs to deal with such errors.
System.out.println("error occurred");
e.printStackTrace();
}
}
}
4 Consume messages
Implementation method
- Create a consumer subscription program ConsumerDemo.java.
- Compile and run ConsumerDemo.java to consume messages.
- check the running results.
- the running result example is as follows.
- Consume: ConsumerRecord(topic = sasl, patition = 6, leaderEpoch=0, offset=0, CreateTime = 1637207092476, serialized key size = -1, serialized key size = 24, headers = ReacordHeaders(headers = [], isReadOnly = false), key = null, value = this is demo, message : 0)Consume: ConsumerRecord(topic = sasl, patition = 6, leaderEpoch=0, offset=0, CreateTime = 1637207092476, serialized key size = -1, serialized key size = 24, headers = ReacordHeaders(headers = [], isReadOnly = false), key = null, value = this is demo, message : 1)
tip
Message Queue for Kafka provides an example Demo for quick access. After downloading and unzipping the Demo, you can run the following command directly to consume messages.
java -cp kafka-demo.jar com.volcengine.openservice.kafka.ConsumerDemo ./config.properties
Example code
The sample code for consuming messages via PLAINTEXT endpoint is as follows. You can also refer to the example file {DemoPath}/src/main/java/com/volcengine/openservice/kafka/ConsumerDemo.java in the Demo to implement related business logic.
package com.volcengine.openservice.kafka;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
class Consumer {
//Startup configuration of consumer
private static Properties props = new Properties();
private KafkaConsumer<String, String> consumer;
//Instance constructor of consumer
public Consumer(Properties kafkaProperties) {
setProps(kafkaProperties);
setSasl(kafkaProperties);
newConsumer();
subscribed(kafkaProperties);
}
// Set the security protocol. Kafka supports SASL_PLAINTEXT and PLAINTEXT protocols
private void setSasl(Properties kafkaProperties) {
String protocol = kafkaProperties.getProperty("security.protocol");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, protocol);
// If the security protocol is PLAINTEXT, no username or password is needed; return directly.
if (protocol.equals("PLAINTEXT")) {
return;
}
// If the security protocol is SASL_PLAINTEXT, obtain the encryption type and configuration of sasl
if (protocol.equals("SASL_PLAINTEXT")) {
props.put(SaslConfigs.SASL_MECHANISM, kafkaProperties.getProperty("sasl.mechanism"));
props.put(SaslConfigs.SASL_JAAS_CONFIG, kafkaProperties.getProperty("sasl.jaas.config"));
return;
}
throw new IllegalArgumentException("security.protocol is not correct");
}
// Set the startup parameters of consumer
private void setProps(Properties kafkaProperties) {
// Set the endpoint. Obtain the endpoint for the corresponding topic from the console.
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
// This value can be set based on actual data polling and client version. The default setting is 30s.
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
// Maximum number of records per poll
// Do not set this value too high. If the polled data is too large that they can not be consumed completely before the next poll, it may trigger a load balancing and cause stutter.
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
// Message deserialization method
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
// The consumer group to which the current consumption instance belongs.
// Instances in the same group share the message load
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("consumer.group.id"));
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaProperties.getProperty("consumer.auto.offset.reset"));
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaProperties.getProperty("consumer.enable.auto.commit"));
//Set the DNS policy of client
props.put(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG, kafkaProperties.getProperty("client.dns.lookup"));
}
// Construct a consumer object, i.e., a consumer instance.
private void newConsumer() {
consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
}
// Subscribe to topic
private void subscribed(Properties kafkaProperties) {
// Set the subscription topic for the consumer group.
List<String> subscribedTopic = new ArrayList<String>();
// If GROUP_ID_CONFIG is the same, it is recommended to set the subscribed topic the same.
// Topic must be created first in the console
subscribedTopic.add(kafkaProperties.getProperty("topic"));
consumer.subscribe(subscribedTopic);
}
// Consume messages
public void poll() {
ConsumerRecords<String, String> records = consumer.poll(1000);
// These data must be consumed before the next poll, and total time must not exceed SESSION_TIMEOUT_MS_CONFIG
// We recommend that you open a separate thread pool to consume messages, and then return responses asynchronously
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Consume: %s%n", record.toString());
}
}
}
public class ConsumerDemo {
public static void main(String args[]) {
if (args.length < 1) {
System.out.println("please input config path");
return;
}
// Get the consumer object
Consumer consumer = new Consumer(KafkaConfigurer.getKafkaProperties(args[0]));
// Consume messages in a loop
while (true){
try {
// Consume messages
consumer.poll();
} catch (Exception e) {
try {
Thread.sleep(1000);
} catch (Throwable ignore) {
}
e.printStackTrace();
}
}
}
}
Last updated: 2025.09.18 17:51:06
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
