一、訊息發送

1.Kafka Java客戶端資料生產流程決議

1. 首先要構造一個 ProducerRecord 物件，該物件可以宣告主題Topic、磁區Partition、鍵 Key以及值 Value，主題和值是必須要宣告的，磁區和鍵可以不用指定，
2. 呼叫send() 方法進行訊息發送，
3. 因為訊息要到網路上進行傳輸，所以必須進行序列化，序列化器的作用就是把訊息的 key 和 value物件序列化成位元組陣列，
4. 接下來資料傳到磁區器，如果之間的 ProducerRecord 物件指定了磁區，那么磁區器將不再做任何事，直接把指定的磁區回傳；如果沒有，那么磁區器會根據 Key 來選擇一個磁區，選擇好磁區之后，生產者就知道該往哪個主題和磁區發送記錄了，
5. 接著這條記錄會被添加到一個記錄批次里面，這個批次里所有的訊息會被發送到相同的主題和磁區，會有一個獨立的執行緒來把這些記錄批次發送到相應的 Broker 上，
6. Broker成功接收到訊息，表示發送成功，回傳訊息的元資料（包括主題和磁區資訊以及記錄在磁區里的偏移量），發送失敗，可以選擇重試或者直接拋出例外，

依賴的包 <kafka.version>2.0.0</kafka.version>

<dependency>
	<groupId>org.apache.kafka</groupId> 
	<artifactId>kafka_${scala.version}</artifactId> 
	<version>${kafka.version}</version> 
	<exclusions> 
		<exclusion> 
			<groupId>org.apache.zookeeper</groupId> 
			<artifactId>zookeeper</artifactId> 
		</exclusion> 
		<exclusion> 
			<groupId>org.slf4j</groupId> 
			<artifactId>slf4j-log4j12</artifactId> 
		</exclusion> 
		<exclusion> 
			<groupId>log4j</groupId> 
			<artifactId>log4j</artifactId> 
		</exclusion> 
	</exclusions> 
</dependency>

2.必要引數配置

見代碼庫：com.heima.kafka.chapter2.KafkaProducerAnalysis

public static Properties initConfig() { 
	Properties props = new Properties(); 
	// 該屬性指定 brokers 的地址清單，格式為 host:port，清單里不需要包含所有的 broker 地址， 
	// 生產者會從給定的 broker 里查找到其它 broker 的資訊，——建議至少提供兩個 broker 的資訊，因為一旦其中一個宕機，生產者仍然能夠連接到集群上， 
	props.put("bootstrap.servers", brokerList); 
	// 將 key 轉換為位元組陣列的配置，必須設定為一個實作了 org.apache.kafka.common.serialization.Serializer 介面的類， 
	// 生產者會用這個類把鍵物件序列化為位元組陣列， 
	// ——kafka 默認提供了 StringSerializer和 IntegerSerializer、 ByteArraySerializer，當然也可以自定義序列化器， 
	props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
	// 和 key.serializer 一樣，用于 value 的序列化 
	props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
	// 用來設定KafkaProducer對應的客戶端ID，默認為空，如果不設定KafkaProducer會自動 生成一個非空字串， 
	// 內容形式如："producer-1" 
	props.put("client.id", "producer.client.id.demo"); 
	return props; 
}

Properties props = initConfig(); 
	KafkaProducer<String, String> producer = new KafkaProducer<>(props); 
// 	KafkaProducer<String, String> producer = new KafkaProducer<>(props, 
// new StringSerializer(), new StringSerializer()); 
//生成 ProducerRecord 物件，并制定 Topic，key 以及 value 
	ProducerRecord<String, String> record = new ProducerRecord<>(topic, "hello, Kafka!"); 
	try {
		// 發送訊息 
		producer.send(record);
	}

3.發送型別

發送即忘記

producer.send(record)

同步發送

//通過send()發送完訊息后回傳一個Future物件,然后呼叫Future物件的get()方法等待kafka回應 
//如果kafka正常回應，回傳一個RecordMetadata物件，該物件存盤訊息的偏移量 
// 如果kafka發生錯誤，無法正常回應，就會拋出例外，我們便可以進行例外處理 
producer.send(record).get();

異步發送

producer.send(record, new Callback() { 
	public void onCompletion(RecordMetadata metadata, Exception exception) { 
	if (exception == null) { 
		System.out.println(metadata.partition() + ":" + metadata.offset()); 
	} 
} 
});

4.序列化器

訊息要到網路上進行傳輸，必須進行序列化，而序列化器的作用就是如此，

Kafka 提供了默認的字串序列化器（org.apache.kafka.common.serialization.StringSerializer），還有整型（IntegerSerializer）和位元組陣列（BytesSerializer）序列化器，這些序列化器都實作了介面（org.apache.kafka.common.serialization.Serializer）基本上能夠滿足大部分場景的需求，

5.自定義序列化器

見代碼庫：com.heima.kafka.chapter2.CompanySerializer

/**
	* 自定義序列化器
	*/
public class CompanySerializer implements Serializer<Company> { 
	@Override 
	public void configure(Map configs, boolean isKey) { 
	}
	
	@Override 
	public byte[] serialize(String topic, Company data) { 
		if (data == null) { 
			return null; 
		}
		byte[] name, address; 
		try {
			if (data.getName() != null) { 
				name = data.getName().getBytes("UTF-8"); 
			} else { 
				name = new byte[0]; 
			}
			if (data.getAddress() != null) { 
				address = data.getAddress().getBytes("UTF-8"); 
			} else { 
				address = new byte[0]; 
			}
			ByteBuffer buffer = ByteBuffer. allocate(4 + 4 + name.length + address.length); 
			buffer.putInt(name.length); 
			buffer.put(name); 
			buffer.putInt(address.length); 
			buffer.put(address); 
			return buffer.array(); 
		} catch (UnsupportedEncodingException e) { 
			e.printStackTrace(); 
		}
		return new byte[0]; 
	}
	
	@Override 
	public void close() { 
	} 
}

• 使用自定義的序列化器

見代碼庫：com.heima.kafka.chapter2.ProducerDefineSerializer

public class ProducerDefineSerializer { 
	public static final String brokerList = "localhost:9092"; 
	public static final String topic = "heima"; 
	
	public static void main(String[] args) throws ExecutionException, InterruptedException { 
		Properties properties = new Properties(); 
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CompanySerializer.class.getName()); 
// properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
// ProtostuffSerializer.class.getName()); 
	properties.put("bootstrap.servers", brokerList); 
	
	KafkaProducer<String, Company> producer = new KafkaProducer<>(properties);
	Company company = Company.builder().name("kafka") .address("北京").build(); 
// Company company = Company.builder().name("hiddenkafka") 
// .address("China").telphone("13000000000").build(); 
	ProducerRecord<String, Company> record = new ProducerRecord<>(topic, company); 
	producer.send(record).get(); 
	} 
}

6.磁區器

本身kafka有自己的磁區策略的，如果未指定，就會使用默認的磁區策略：

Kafka根據傳遞訊息的key來進行磁區的分配，即hash(key) % numPartitions，如果Key相同的話，那么就會分配到統一磁區，

源代碼org.apache.kafka.clients.producer.internals.DefaultPartitioner分析

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 
	List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); 
	int numPartitions = partitions.size(); 
	if (keyBytes == null) { 
		int nextValue = this.nextValue(topic); 
		List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); 
		if (availablePartitions.size() > 0) { 
			int part = Utils.toPositive(nextValue) % availablePartitions.size(); 
			return ((PartitionInfo)availablePartitions.get(part)).partition(); 
		} else { 
			return Utils.toPositive(nextValue) % numPartitions; 
		} 
	} else { 
		return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; 
	} 
}

• 自定義磁區器見代碼庫 com.heima.kafka.chapter2.DefinePartitioner

/**
	* 自定義磁區器 
	*/ 
public class DefinePartitioner implements Partitioner { 
	private final AtomicInteger counter = new AtomicInteger(0); 
	
	@Override 
	public int partition(String topic, Object key, byte[] keyBytes,Object value, byte[] valueBytes, Cluster cluster) { 
	List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); 
	int numPartitions = partitions.size(); 
	if (null == keyBytes) { 
		return counter.getAndIncrement() % numPartitions; 
		} else return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; 
	}
	@Override 
	public void close() { 
	}
	@Override 
	public void configure(Map<String, ?> configs) { 
	} 
}

• 實作自定義磁區器需要通過配置引數ProducerConfig.PARTITIONER_CLASS_CONFIG來實作

// 自定義磁區器的使用 
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,DefinePartitioner.class.getNam e());

7.攔截器

Producer攔截器(interceptor)是個相當新的功能，它和consumer端interceptor是在Kafka 0.10版本被引入的，主要用于實作clients端的定制化控制邏輯，

生產者攔截器可以用在訊息發送前做一些準備作業，

使用場景

1. 按照某個規則過濾掉不符合要求的訊息
2. 修改訊息的內容
3. 統計類需求

見代碼庫：自定義攔截器com.heima.kafka.chapter2.ProducerInterceptorPrefix

/**
	* 自定義攔截器 
	*/ 
public class ProducerInterceptorPrefix implements ProducerInterceptor<String, String> { 
	private volatile long sendSuccess = 0; 
	private volatile long sendFailure = 0; 
	
	@Override 
	public ProducerRecord<String, String> onSend( ProducerRecord<String, String> record) { 
		String modifiedValue = "prefix1-" + record.value(); 
		return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), record.key(), modifiedValue, record.headers()); 
// if (record.value().length() < 5) { 
// throw new RuntimeException();
// } 
// return record; 
	}
	
	@Override 
	public void onAcknowledgement( RecordMetadata recordMetadata, Exception e) { 
		if (e == null) { 
			sendSuccess++; 
		} else { 
			sendFailure++; 
		} 
	}
	
	@Override 
	public void close() { 
		double successRatio = (double) sendSuccess / (sendFailure + sendSuccess); 
		System.out.println("[INFO] 發送成功率=" + String.format("%f", successRatio * 100) + "%"); 
	}
	
	@Override 
	public void configure(Map<String, ?> map) { 
	} 
}

• 實作自定義攔截器之后需要在配置引數中指定這個攔截器，此引數的默認值為空，如下：

// 自定義攔截器使用 
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,ProducerDefineSerializer.cla ss.getName());

• 功能演示：

發送端

接收端

參考文章《Kafka技術手冊》
需要的同學可加助理VX：C18173184271 備注：CSDN Java_Caiyo 免費獲取Java資料！