Kafka消息發(fā)送

1. 構(gòu)建ProducerRecord對象

該類如下：

public class ProducerRecord<K, V> {

    //The topic the record will be appended to
    private final String topic;
    //The partition to which the record should be sent
    private final Integer partition;
    //the headers that will be included in the record
    private final Headers headers;
    //The key that will be included in the record
    private final K key;
    //The record contents
    private final V value;
    //The timestamp of the record, in milliseconds since epoch. If null, the producer will assign the timestamp using System.currentTimeMillis().
    private final Long timestamp;
    
    public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {
        if (topic == null)
            throw new IllegalArgumentException("Topic cannot be null.");
        if (timestamp != null && timestamp < 0)
            throw new IllegalArgumentException(
                    String.format("Invalid timestamp: %d. Timestamp should always be non-negative or null.", timestamp));
        if (partition != null && partition < 0)
            throw new IllegalArgumentException(
                    String.format("Invalid partition: %d. Partition number should always be non-negative or null.", partition));
        this.topic = topic;
        this.partition = partition;
        this.key = key;
        this.value = value;
        this.timestamp = timestamp;
        this.headers = new RecordHeaders(headers);
    }
    
    public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {
        this(topic, partition, timestamp, key, value, null);
    }
    
    public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {
        this(topic, partition, null, key, value, headers);
    }
    
    public ProducerRecord(String topic, Integer partition, K key, V value) {
        this(topic, partition, null, key, value, null);
    }
    
    public ProducerRecord(String topic, K key, V value) {
        this(topic, null, null, key, value, null);
    }
    
    //Create a record with no key
    public ProducerRecord(String topic, V value) {
        this(topic, null, null, null, value, null);
    }
    //省略getters和setters
}

這里有5個構(gòu)造方法，但是最后用的就是其中的一個。實際應(yīng)用中，構(gòu)建ProducerRecord對象是非常頻繁的操作。

2. 發(fā)送消息

發(fā)送消息有三種模式，發(fā)后即忘(fire-and-forget), 同步(sync),異步(async)

2.1 fire-and-forget（上一篇博客介紹的就是發(fā)后即忘）

 producer.send(record);

特點： 效率高，可靠性差

2.2 同步

代碼如下：

package com.ghq.kafka.server;

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;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 消息的發(fā)送
 */
public class ProducerSendMessage {

    public static final String brokerList = "192.168.52.135:9092";
    public static final String topic = "topic-demo";

    public static Properties initProperties(){
        Properties prop = new Properties();
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        prop.put(ProducerConfig.CLIENT_ID_CONFIG,"producer.client.id.demo");
        /**
         *
         * 配置重試次數(shù)，這里重試10次，重試10次之后如果消息還是發(fā)送不成功，那么還是會拋出異常
         * 那些類可以重試呢？
         * org.apache.kafka.common.errors.RetriableException 及其子類
         *
         */
        prop.put(ProducerConfig.RETRIES_CONFIG,10);
        return prop;
    }

    public static void sync() {

        Properties prop = initProperties();
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, "Hello,World");

        //3. 發(fā)送消息
        while (true){
            /**
             * send方法本身就是異步的
             */
            Future<RecordMetadata> future = producer.send(record);

            try {
                /**
                 * get方法是阻塞的
                 * 這里返回 RecordMetadata，包含了發(fā)送消息的元數(shù)據(jù)信息
                 */
                RecordMetadata metadata = future.get();
                System.out.println("topic:"+metadata.topic());
                System.out.println("partition:"+metadata.partition());
                System.out.println("offset:"+metadata.offset());
                System.out.println("hasTimestamp:"+metadata.hasTimestamp());
                System.out.println("-----------------------------------------");
                Thread.sleep(1000);
            } catch (ExecutionException | InterruptedException e) {
                e.printStackTrace();
            }

        }
        //4. 關(guān)閉資源
        //producer.close();

    }
}

特點：性能差，可靠性高

注：什么異?？梢灾卦嚕縍etriableException

RetriableException.jpg

2.3 異步

public static void async() {
        Properties prop = initProperties();
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, "Hello,World");

        Future<RecordMetadata> future = producer.send(record, new Callback() {

            /**
             * metadata 和 exception 互斥
             * 消息發(fā)送成功：metadata != null exception == null
             * 消息發(fā)送失?。簃etadata == null exception != null
             */
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {

                if (exception != null) {
                    System.out.println("消息發(fā)送失敗："+metadata);
                }else {
                    System.out.println("消息發(fā)送成功："+metadata);
                }
            }
        });
        
        //這里采用lambda表達式
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                System.out.println("消息2發(fā)送失?。?+metadata);
            }else {
                System.out.println("消息2發(fā)送成功："+metadata);
            }
        });
        producer.close();
    }

輸出結(jié)果如下：

消息1發(fā)送成功：topic-demo-0@22
消息2發(fā)送成功：topic-demo-2@24

特點：性能 ：同步 < 異步 < 發(fā)后即忘，可靠性：同步 > 異步 > 發(fā)后即忘

結(jié)束。