The key to producing and consuming Apache Kafka messages using Java is to properly configure the Producer and Consumer APIs and understand their basic processes. 1. First add Kafka client dependencies to ensure that the version is compatible with the cluster; 2. When writing producers, configure bootstrap.servers, key.serializer and value.serializer, and create a KafkaProducer instance to send messages, pay attention to closing resources and optional callback processing; 3. When writing consumers, configure group.id, deserializer, etc., use KafkaConsumer to subscribe to topics and pull messages in a loop, pay attention to the strategy of submitting offsets; 4. Common problems include group.id conflicts, inappropriate submission of offsets, and incorrect resource closing, etc. It is recommended to test the verification process in a local stand-alone environment first.

It is actually not difficult to produce and consume Apache Kafka messages in Java. The core lies in understanding the basic concepts of Kafka and mastering the use of Producer and Consumer APIs. As long as the configuration is properly and the code structure is clear, the sending and receiving of messages can be easily achieved.

Add Kafka dependencies

Before you start writing code, you need to introduce Kafka's client library into your project. If you are using Maven, you can add the following dependencies in pom.xml :

 <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

This version is relatively new and relatively stable. Of course, you can also select the corresponding client version based on your Kafka cluster version.

Write a Kafka Producer

The producer's task is to send messages to the Kafka topic. The main steps include configuring properties, creating a Producer instance, building a message record and sending.

Key configuration items are:

• bootstrap.servers : Kafka broker address
• key.serializer and value.serializer : Specify the serialization method of key and value. StringSerializer is generally used.

The sample code is as follows:

 Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");

producer.send(record);
producer.close();

Some points to note:

• It is best to call close() after sending to avoid resource leakage
• If you want to confirm whether the sending is successful, you can add the callback function.send .send(record, callback)
• If it's just a test, you can omit the key

Write Kafka Consumer

The consumer is responsible for reading messages from the Kafka topic. Compared with producers, the consumer's logic is a little more complicated because it requires actively pulling messages, processing offsets, etc.

Key configurations include:

• bootstrap.servers : also specify the broker address
• group.id : consumer group ID, must be set, otherwise an error will be reported
• key.deserializer and value.deserializer : Deserializer, usually using StringDeserializer

A simple consumer process is as follows:

 Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("Received: " record.value());
    }
}

Note:

• poll() method will pull a batch of data and process it in a loop
• Don’t forget to submit offset after consumption, you can choose to submit automatically or manually
• If you want to consume only once and then exit, you can break loop after processing

Frequently Asked Questions and Precautions

Sometimes you will find that the message has not been received or has been consumed repeatedly, which is usually caused by the following reasons:

• Consumer group.id setting error or conflict
• Automatically submit offset frequency is too high or too low
• Failure to close the producer or consumer correctly results in inconsistent status
• Kafka broker is not started or the network is not connected

It is recommended to first run the local stand-alone environment during development and then go to the cluster.

Basically that's it. The process of connecting to Kafka in Java is not complicated, but some details are easy to ignore, such as serialization, group.id, and submission strategies. As long as you pay attention to the configuration and process, you can run quickly.

The above is the detailed content of How to produce and consume messages from Apache Kafka with Java?. For more information, please follow other related articles on the PHP Chinese website!