Spring-Kafka 2.0.0发送API翻译

简介: Kafka Template–2.2.0 apiKafkaTemplateKafkaTemplate这个类包装了个生产者,来提供方便的发送数据到kafka的topic里面。

Kafka Template–2.2.0 api

KafkaTemplate

KafkaTemplate这个类包装了个生产者,来提供方便的发送数据到kafka的topic里面。
同步和异步的方法都有,异步方法返回一个Future

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(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);

}
  • sendDefault 这个方法,需要提供一个默认的topic。
  • API使用一个timestamp作为一个参数,这个时间戳奖杯存储在record里面。
  • 用户的时间戳是否被存储,取决于在kafka的topic里面配置的时间戳类型。如果topic被配置成了CREATE_TIME,则将使用用户指定的时间戳,如果没指定,将自动生成。如果设置为了LOG_APPEND_TIME那么用户指定的时间戳将会被忽略,使用broker的local time.
  • metrics方法和partitionsFor方法,委托给一些潜在的生产者;而execute方法则是提供了直接访问潜在生产者的途径。
    为了使用这个template,配置个工厂类,并将这个工厂类提供给templage类的构造方法。
    举个栗子:
@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");
    ...
    return props;
}

@Bean
public KafkaTemplate<Integer, String> kafkaTemplate() {
    return new KafkaTemplate<Integer, String>(producerFactory());
}
  • 这个template可以使用spring方式进行配置。
  • 当使用一个Message<?>参数的时候,需要为这个消息的头部,提供topic,partition,key信息。

    • KafkaHeaders.TOPIC
    • KafkaHeaders.PARTITION_ID
    • KafkaHeaders.MESSAGE_KEY
    • KafkaHeaders.TIMESTAMP
      这些信息将被装在到数据里面。
  • 你可以配置给KafkaTemplate配置个ProducerListener来监听异步的回调结果,用这种方式替代使用Future

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();

}
  • template默认配置LoggingProducerListener这个类作为监听类用来记录错误,但是当发送成功时候,却不做任何操作。
  • 为了方便使用,提供了抽象类ProducerListenerAdapter,你可以只实现它一两个方法,它为isInterestedInSuccess方法返回false
  • 观察template类的发送方法,发现它返回一个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这个类有两个属性,ProducerRecordRecordMetadata;具体参考kafka的API
  • 如果你想要阻塞发送线程(批量发送),为了等待结果,你可以调用future的get()方法;另外,在等待的过程中,你可能想要调用flush()方法,为了简便,template类提供了一个autoFlush,这个属性将会在每次send的时候,立即去flush掉sending thread.不过autoFlush将会明显降低性能(就是说最好不要一条条的发送,要批量发送)。
    原文地址:http://docs.spring.io/spring-kafka/docs/2.0.0.M2/reference/htmlsingle/#_sending_messages
目录
相关文章
|
15天前
|
监控 Java 应用服务中间件
高级java面试---spring.factories文件的解析源码API机制
【11月更文挑战第20天】Spring Boot是一个用于快速构建基于Spring框架的应用程序的开源框架。它通过自动配置、起步依赖和内嵌服务器等特性,极大地简化了Spring应用的开发和部署过程。本文将深入探讨Spring Boot的背景历史、业务场景、功能点以及底层原理,并通过Java代码手写模拟Spring Boot的启动过程,特别是spring.factories文件的解析源码API机制。
45 2
|
2月前
|
Java API 数据库
如何使用Spring Boot构建RESTful API,以在线图书管理系统为例
【10月更文挑战第9天】本文介绍了如何使用Spring Boot构建RESTful API,以在线图书管理系统为例,从项目搭建、实体类定义、数据访问层创建、业务逻辑处理到RESTful API的实现,详细展示了每个步骤。通过Spring Boot的简洁配置和强大功能,开发者可以高效地开发出功能完备、易于维护的Web应用。
63 3
|
2月前
|
IDE Java API
基于Spring Boot REST API设计指南
【10月更文挑战第4天】 在现代的软件开发中,RESTful API已经成为了构建网络应用的标准之一。它通过HTTP协议提供了与资源交互的方式,使得不同的应用程序能够进行数据交互。Spring Boot作为一个功能强大的框架,它简化了配置和开发流程,成为了构建RESTful API的理想选择。本文将详细介绍如何在Spring Boot中设计和实现高质量的RESTful API,并提供一些最佳实践。
50 1
|
2月前
|
缓存 Java API
基于Spring Boot REST API设计指南
【10月更文挑战第11天】 在构建现代Web应用程序时,RESTful API已成为一种标准,使得不同的应用程序能够通过HTTP协议进行通信,实现资源的创建、读取、更新和删除等操作。Spring Boot作为一个功能强大的框架,能够轻松创建RESTful API。本文将详细介绍如何在Spring Boot中设计和实现高质量的RESTful API。
124 61
|
2月前
|
存储 安全 Java
jdk21的外部函数和内存API(MemorySegment)(官方翻译)
本文介绍了JDK 21中引入的外部函数和内存API(MemorySegment),这些API使得Java程序能够更安全、高效地与JVM外部的代码和数据进行互操作,包括调用外部函数、访问外部内存,以及使用不同的Arena竞技场来分配和管理MemorySegment。
52 1
jdk21的外部函数和内存API(MemorySegment)(官方翻译)
|
2月前
|
消息中间件 NoSQL Kafka
大数据-52 Kafka 基础概念和基本架构 核心API介绍 应用场景等
大数据-52 Kafka 基础概念和基本架构 核心API介绍 应用场景等
65 5
|
4月前
|
消息中间件 Kafka API
【Kafka消费新风潮】告别复杂,迎接简洁之美——深度解析Kafka新旧消费者API大比拼!
【8月更文挑战第24天】Apache Kafka作为一个领先的分布式流处理平台,广泛用于实时数据管道和流式应用的构建。随着其发展,消费者API经历了重大更新。旧消费者API(包括“低级”和“高级”API)虽提供灵活性但在消息顺序处理上存在挑战。2017年引入的新消费者API简化了接口,自动管理偏移量,支持更强大的消费组功能,显著降低了开发复杂度。通过对比新旧消费者API的代码示例可以看出,新API极大提高了开发效率和系统可维护性。
133 58
|
2月前
|
安全 Java API
基于Spring Boot REST API设计指南
【10月更文挑战第10天】 在现代Web应用开发中,RESTful API扮演着至关重要的角色。Spring Boot作为一个高效、便捷的Java开发框架,为构建RESTful API提供了强大的支持。本文将分享基于Spring Boot的REST API设计指南,涵盖从项目初始化到API文档配置的全过程。
48 0
|
2月前
|
IDE API 定位技术
Python--API编程:IP地址翻译成实际的物理地址
Python--API编程:IP地址翻译成实际的物理地址
|
2月前
|
Java API Maven
使用 Smart-doc 记录 Spring REST API
使用 Smart-doc 记录 Spring REST API
56 0