Author：趙志乾
Date：2018-10-21
Declaration：All Right Reserved！！！

ProducerRecord.java

該類的例項用於存放生產者向kafka叢集傳送的單條訊息記錄。其內容如下，共包含6個屬性欄位。其中的topic欄位用於指定該訊息記錄的一級分屬，partition欄位和key欄位用於指定該訊息記錄的二級分屬，value欄位用於存放該訊息記錄的有效載荷，timestamp欄位用於指定時間戳，headers欄位用於指定.......。

topic欄位指定訊息記錄的一級分屬：kafka叢集可以作為訊息匯流排來使用，此時叢集可以接收位於不同系統中的生產者傳送來的不同類別的訊息，而topic欄位便可認為是邏輯層面的一種劃分。

partition欄位和key欄位指定訊息記錄的二級分屬：kafka叢集為了提高訊息投遞和消費的併發度，引入了分割槽概念，對訊息記錄進行二級分屬指定。這兩個欄位屬於可選欄位，不同的組合方式對應不同的劃分策略。

partition和key欄位決定的劃分策略：
    1、如果partiton欄位不空，則直接由該欄位指定具體分割槽，實現分割槽的定投；
    2、如果partition欄位為空，且key欄位不為空，則通過對key值進行hash，來確定訊息記錄應歸屬的二級歸屬；
    3、如果partition和key欄位均為空，則對分割槽進行輪詢投遞，實現負載均衡；

value欄位存放訊息記錄有效載荷：生產者真正要釋出的訊息內容會存放於value欄位；

timestamp欄位指定時間戳：如果使用者沒有指定該欄位，則其會有生產者自動填充。而其最終的值是由topic中指定的時間戳型別來決定的，如果指定為建立時間，則該值由生產者客戶端填寫；如果指定為日誌追加時間，則該值會由broker在將訊息記錄持久化後用當前時間進行覆蓋。

注：在構造該類的例項時，要保證主題不能為空，分割槽號如果設定的話，要保證其值不小於0的整數，而時間戳如果設定的話，也要保證其值為不小於0的整數。

package org.apache.kafka.clients.producer;

import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;

/*該類的例項用於存放生產者要向kafka叢集釋出的訊息記錄，其包括一個topic名稱，用來給訊息記錄做邏輯層面的一級劃分，即該訊息記錄應該被髮布到kafka叢集的哪個類別（主題）下；一個可選的分割槽號，用來給訊息記錄做邏輯上二級劃分，即該訊息記錄應該被髮布到kafka叢集指定類別下的哪個佇列（分割槽）中；*/
public class ProducerRecord<K, V> {

    //訊息主題
    private final String topic;
    //分割槽號：可選
    private final Integer partition;
    private final Headers headers;
    //key值：可選
    private final K key;
    //value值：生產者要傳送的有效訊息載荷
    private final V value;
    //時間戳
    private final Long timestamp;

    //建立一個訊息記錄例項，所需指定的引數包括：主題、分割槽號、時間戳、key、value和headers
    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);
    }


    //建立一個訊息記錄例項，所需指定的引數包括：主題、分割槽號、時間戳、key、value
    public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {
        this(topic, partition, timestamp, key, value, null);
    }

    //建立一個訊息記錄例項，所需指定的引數包括：主題、分割槽號、key、value、headers
    public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {
        this(topic, partition, null, key, value, headers);
    }
    
    //建立一個訊息記錄例項，所需指定的引數包括：主題、分割槽號、key值和訊息有效載荷value值
    public ProducerRecord(String topic, Integer partition, K key, V value) {
        this(topic, partition, null, key, value, null);
    }
    
    //建立一個訊息記錄例項，所需指定的引數包括主題、key值和訊息有效載荷value值
    public ProducerRecord(String topic, K key, V value) {
        this(topic, null, null, key, value, null);
    }
    
    //建立一個訊息記錄例項，所需指定的引數包括：主題和訊息有效載荷value值
    public ProducerRecord(String topic, V value) {
        this(topic, null, null, null, value, null);
    }

    //返回記錄的主題名稱
    public String topic() {
        return topic;
    }

    public Headers headers() {
        return headers;
    }

    //返回訊息記錄的key值
    public K key() {
        return key;
    }

    //返回訊息記錄的value值
    public V value() {
        return value;
    }

    //返回訊息記錄的時間戳：單位毫秒
    public Long timestamp() {
        return timestamp;
    }

    //返回訊息記錄要被投遞到的分割槽號
    public Integer partition() {
        return partition;
    }

    @Override
    public String toString() {
        String headers = this.headers == null ? "null" : this.headers.toString();
        String key = this.key == null ? "null" : this.key.toString();
        String value = this.value == null ? "null" : this.value.toString();
        String timestamp = this.timestamp == null ? "null" : this.timestamp.toString();
        return "ProducerRecord(topic=" + topic + ", partition=" + partition + ", headers=" + headers + ", key=" + key + ", value=" + value +
            ", timestamp=" + timestamp + ")";
    }

    @Override
    public boolean equals(Object o) {
        if (this == o)
            return true;
        else if (!(o instanceof ProducerRecord))
            return false;

        ProducerRecord<?, ?> that = (ProducerRecord<?, ?>) o;

        if (key != null ? !key.equals(that.key) : that.key != null) 
            return false;
        else if (partition != null ? !partition.equals(that.partition) : that.partition != null) 
            return false;
        else if (topic != null ? !topic.equals(that.topic) : that.topic != null) 
            return false;
        else if (headers != null ? !headers.equals(that.headers) : that.headers != null)
            return false;
        else if (value != null ? !value.equals(that.value) : that.value != null) 
            return false;
        else if (timestamp != null ? !timestamp.equals(that.timestamp) : that.timestamp != null)
            return false;

        return true;
    }

    @Override
    public int hashCode() {
        int result = topic != null ? topic.hashCode() : 0;
        result = 31 * result + (partition != null ? partition.hashCode() : 0);
        result = 31 * result + (headers != null ? headers.hashCode() : 0);
        result = 31 * result + (key != null ? key.hashCode() : 0);
        result = 31 * result + (value != null ? value.hashCode() : 0);
        result = 31 * result + (timestamp != null ? timestamp.hashCode() : 0);
        return result;
    }
}