kafka生产者总结(五)

就今天我对kafka的生产做个总结，首先我们来看一下我们来看看通过java实现kafka生产者客户端向kafka集群发送消息的通用代码：

public class Producer extends Thread {
private final KafkaProducer<Integer, String> producer;
private final String topic;
private final Boolean isAsync;

public Producer(String topic, Boolean isAsync) {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producer = new KafkaProducer<>(props);
this.topic = topic;
this.isAsync = isAsync;
}
public void run() {
int messageNo = 10;
while (true) {
String messageStr = "Message_" + messageNo;
long startTime = System.currentTimeMillis();
if (isAsync) { // Send asynchronously
producer.send(new ProducerRecord<>(topic,
messageNo,
messageStr), new DemoCallBack(startTime, messageNo, messageStr));
} else { // Send synchronously
try {
producer.send(new ProducerRecord<>(topic,
messageNo,
messageStr)).get();
System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")");
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
++messageNo;
}
}
}
class DemoCallBack implements Callback {
private final long startTime;
private final int key;
private final String message;

public DemoCallBack(long startTime, int key, String message) {
this.startTime = startTime;
this.key = key;
this.message = message;
}
public void onCompletion(RecordMetadata metadata, Exception exception) {
long elapsedTime = System.currentTimeMillis() - startTime;
if (metadata != null) {
System.out.println(
"message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
"), " +
"offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
} else {
exception.printStackTrace();
}
}

}

向kafka集群生产消息分为同步和异步两种方式，同步方式即同步等到消息发送完成再返回结果。异步方式是传入一个回调方法，在消息发送完成时异步回调找个方法。kafka生产者线程安全的，如果在多线程的情况下生产消息，建议多个线程之间公用一个KafkaProducer对象，而不是每个线程new 一个KafkaProducer对象。

kafka生产消息的大致流程如下图：

kafka生产者总结(五)

通过上面的流程可以看到，kafka生产者发送的消息实际上是换到到了RecordAccumulator对象中，内部是通过一个map缓存消息。后台起了一个Sender守护线程，根据一定的条件从RecordAccumulator对象中取出消息通过NIO网络请求发送给kafaka集群，这样设计的目的主要是增加生产消息的处理效率，当然这其中还涉及到很多细节，有兴趣的话可以阅读这一块的代码。

我们知道kafka的topic是分区的，每个分区又有多个副本，只有leader副本负责处理读写消息的请求，follower副本负责从leader副本同步消息。对于生产者有一个比较重要的参数是acks，它表示必须有多少个分区副本收到消息生产者才认为消息是写入成功的，acks=0，表示生产者写入消息不会等待服务器的任何响应，这个方式可能造成消息的丢失，但能提高吞吐量

acks=1，表示只有leader副本收到消息，生产者就会收到一个来自服务器成功的响应，acks=-1，只有当所有的复制节点收到消息时，生产者才会收到成功的响应。

acks参数需要参考系统的吞吐量以及消息的可靠性来设置

kafka生产者总结(五)

相关推荐