Common asynchronous sending code writing

Catalog

Asynchronous sending code writing

  Asynchronous sending code with callback function

  The synchronous API Code writing

Asynchronous sending code writing

1、 establish Kafka engineering

2、 Import dependence

The contents are as follows ：

<dependencies>

                <dependency>

                           <groupId>org.apache.kafka</ groudId>

                           <artifactId>kafka-clients</ artifactId>

                            <version>3.0.0</ version>

                <dependency>

</ dependencies>

3、 Create a package ：com.atguigu.kafka.producer

Write the code as follows ：

package com.atguigu.kafka.producer;
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.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducer {

    public static void main(String[] args) {

        // 0 To configure 
        Properties properties = new Properties();

        //  Connect clusters  bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");

        //  Specify corresponding key and value The serialization type of 
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 1 establish kafka Producer object 

        // “”hello
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);


        // 2 send data 
        for (int i=0; i<5; i++) {

            kafkaProducer.send(new ProducerRecord<>("first","atguigu"+i));

        }
        // 3 close resource 
        kafkaProducer.close();
    }



}

4、 Preparation for the test

1) stay Hadoop103 Upper opening Kafka consumer

[[email protected] kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

2） stay IDEA Execute code in , Observe hadoop103  Whether a message is received in the console .

As shown in the figure, it is correct

  Asynchronous sending code with callback function

The callback function will be in producer received ack Called when the , Called asynchronously , This method has two parameters , Metadata information （RecordMetadata） And exception information （Exception）, If Exception by null, Explain that the message is sent Send successfully , If Exception Not for null, Message sending failed .

Be careful ： If the message fails to be sent, it will be automatically retried , We don't need to manually try again in the callback function .

Ctrl+C   +   Ctrl+V

You can recreate a package name ：CustomProducerCallback

Code writing ：

Just change for The corresponding code under the statement

  Preparation for the test

1） stay Hadoop103 Upper opening Kafka consumer

[[email protected] kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

2） stay IDEA Execute code in , Observe hadoop103  Whether the following message is received in the console .

 3） stay IDEA Console observation callback information .

  The synchronous API Code writing

  The synchronous API The code is written as follows

Ctrl+C   +   Ctrl+V

You can recreate a package name ：CustomProducerSync

Code writing ：

Just change for The corresponding code under the statement

Just send it asynchronously , Call it again get() The method can .

  And pay attention to him get Category

  Preparation for the test

1) stay Hadoop103 Upper opening Kafka consumer

[[email protected] kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

2、 stay IDEA Execute code in , Observe hadoop103  Whether a message is received in the console .