使用KafkaTemplate发送消息
定义：
KafkaTemplate包装了一个生产者，并提供方便的方法用于发送数据给Kafka topic。
使用KafkaTemplate方法前需要配置一个Producer的工厂类，并将它作为Kafkatemplate的构造器参数传入构造实例。
@Bean
public ProducerFactory<Integer, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// See https://kafka.apache.org/documentation/#producerconfigs for more properties
return props;
}
@Bean
public KafkaTemplate<Integer, String> kafkaTemplate() {
return new KafkaTemplate<Integer, String>(producerFactory());
}
可用API以及说明：
ListenableFuture<SendResult<K, V>> sendDefault(V data);
ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, V data);
ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
ListenableFuture<SendResult<K, V>> send(Message<?> message);
Map<MetricName, ? extends Metric> metrics();
List<PartitionInfo> partitionsFor(String topic);
<T> T execute(ProducerCallback<K, V, T> callback);
// Flush the producer.
void flush();
interface ProducerCallback<K, V, T> {T doInKafka(Producer<K, V> producer);}
说明：
1、 sendDefault方法使用前需要配置一个默认的 topic给template；
2、 方法中入参含有timestamp时间戳的，会将给时间戳存储在记录中；
3、 metrics和partitionsFor方法对应Producer类的同名方法。execute方法提供了直接访问Producer的能力；
4、 当使用一个方法入参含有Message<?>参数是，topic、partition还有其他关键信息都可以在这个message header中获得；
• KafkaHeaders.TOPIC
• KafkaHeaders.PARTITION_ID
• KafkaHeaders.MESSAGE_KEY
• KafkaHeaders.TIMESTAMP
5、 如果想要获取异步的回调函数信息，其中带有成功以及失败的信息的话，可以配置KafkaTemplate以及ProducerListener；
public interface ProducerListener<K, V> {
void onSuccess(String topic, Integer partition, K key, V value, RecordMetadata recordMetadata);
void onError(String topic, Integer partition, K key, V value, Exception exception);
boolean isInterestedInSuccess();
}
6、 默认情况下，Template配置了 LoggingProducerListenner，该类只记录错误信息，发送成功信息不做记录；
7、 onSuccess只当isInterestedInsuccess返回True时被调用；
8、 出于方便性考虑，当你只想实现其中的一种方法是，抽象类ProducerListenerAdapter可以实现该需求。对于 isInterestedInSuccess它会返回false值；
9、 发送数据后返回ListenableFuture<SendResult>，可以注册一个回调函数去异步接收发送结果数据信息；
ListenableFuture<SendResult<Integer, String>> future = template.send("foo");
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
@Override
public void onSuccess(SendResult<Integer, String> result) {
//成功代码处理逻辑
}
@Override
public void onFailure(Throwable ex) {
//失败代码处理逻辑
}
});
SendResult有两个属性，分别为 ProducerRecord以及RecordMetaData。
如果想要同步获取发送结果数据，可以通过future的get方法获取；
template中存在一个autoFlash构造函数，该参数设为true可以导致每次发送都会调用flush（）方法，可能降低程序性能。
代码实例：
异步方法：
public void sendToKafka(final MyOutputData data) {
final ProducerRecord<String, String> record = createRecord(data);
ListenableFuture<SendResult<Integer, String>> future = template.send(record);
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
@Override
public void onSuccess(SendResult<Integer, String> result) {
handleSuccess(data);
}
@Override
public void onFailure(Throwable ex) {
handleFailure(data, record, ex);
}
});
}
同步方法：
public void sendToKafka(final MyOutputData data) {
final ProducerRecord<String, String> record = createRecord(data);
try {
template.send(record).get(10, TimeUnit.SECONDS);
handleSuccess(data);
}
catch (ExecutionException e) {
handleFailure(data, record, e.getCause());
}
catch (TimeoutException | InterruptedException e) {
handleFailure(data, record, e);
}
}
