Kafka - 生产者

生产者消息对象

public class ProducerRecord<K, V> {private final String topic; // 主题private final Integer partition; //分区号private final Headers headers; //消息头部private final K key; //键private final V value; //值private final Long timestamp; //消息的时间戳
}

其中key是用来指定消息的键，它不仅是消息的附加信息，还可以用来计算分区号，进而让消息发往特定的分区，一般同一个key的消息会被划分到同一个分区中。
timestamp是指消息的时间戳，它有CreateTime和LogAppendTime两种类型，前者表示消息创建的时间，后者表示消息追加到日志文件的时间。

创建生产者实例

public static Properties initConfig() {Properties props = new Properties();props.put(ProducerConfig.KEY_SERIALZER_CLASS_CONFIG,StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALZER_CLASS_CONFIG,StringSerializer.class.getName());
}KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());

消息的发送

创建生产者实例

创建生产者实例的方法有很多种，其中最简单的是下面的构造方于除了topic和value外的属性，其他都置为null。

public ProducerRecord(String topic, V value);

发送消息主要有三种模式：发完即忘（fire-and-forget），同步（sync）及异步（async）。

KafkaProducer的sand()方法返回值并非是void类型，而是Future类型，send()方法有两个重载方法，具体定义如下：

public Future<RecordMetadata> send(ProducerRecord<K,V> record);
public Future<RecordMetadata> send(ProducerRecord<K,V> record, Callback callback);

发完即忘
它只管往Kafka中发送消息而并不关心消息是否正确到达。
在大多数情况下，这种发送方式没有什么问题，不过在某些时候（比如发生不可重试异常时），会造成消息的丢失。这种发送方式性能最高，但可靠性也最差。
同步发送

try {producer.send(record).get();
} catch (ExecutionException | InterruptedException e) {e.printStackTrace();
}

通过feature对象中的get()方法，来阻塞等待kafka的响应，直到发送成功，或者发生异常。

同步发送的可靠性高，但性能会差很多，因为需要阻塞等到一条消息发送完之后，才能发送下一条。

异步发送

producer.send(record, new Callback()) {@overridepublic void onCompletion(RecordMetadata metadata, Excetion exception){if (excetion != null) {exception.printStackTrace();} else {...}}
}

当Kafka有响应时候，就会有回调，要么发送成功，要么抛出异常。

序列化器

生产者需要用序列化器把对象转换成字节数组，才能通过网络发送给Kafka。而消费者需要用反序列器把从Kafka中收到的字节数组转换成相应的对象。

分区器

分区器的作用是为消息分配分区。

消息经过序列化后，就需要确定它发往的分区，如果消息ProducerRecord中指定了partition字段，那么就不需要分区器，因为patition代表的就是要发往的分区号。如果没有指定partition，则需要依赖分区器，根据key字段来计算partition的值。

拦截器

生产者拦截器既可以用来在消息发送前做一些准备工作，比如按照某个规则过滤不符合要求的消息、修改消息的内容等，也可以用来在发送回调逻辑前，做一些定制化的需求，比如统计类工作。

原理分析

整体架构

在这里插入图片描述
整个生产者客户端由两个线程协调运行，这两个线程分别为主线程和Sender线程。在主线程中由KafkaProducer创建消息，然后通过可能的拦截器、序列化器和分区器的作用后，缓存到消息收集器中（RecordAccumulator）。Sender现成负责从消息收集器中获取消息，并将其发送到kafka中。