1. 参考

这部分参考文档详细介绍了构成Spring for Apache Kafka的各种组件。该主章涵盖了核心类开发一个带有Spring 的Kafka 应用程序。

1.1 使用Spring for Apache Kafka

本节详细解释了使用Spring对Apache Kafka产生影响的各种问题。

1.1.1 配置Topic

如果我们在应用程序上下文中定义了一个KafkaAdmin bean，它将会自动向broker(代理)添加Topic。为此，我们可以为应用程序上下文中的每个主题添加一个 NewTopic @Bean。以下示例显示了如何执行此操作：

@Bean
public KafkaAdmin admin() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
            StringUtils.arrayToCommaDelimitedString(embeddedKafka().getBrokerAddresses()));
    return new KafkaAdmin(configs);
}

@Bean
public NewTopic topic1() {
    return new NewTopic("thing1", 10, (short) 2);
}

@Bean
public NewTopic topic2() {
    return new NewTopic("thing2", 10, (short) 2);
}

默认情况下，如果broker(代理)不可用，则会记录一条消息，但会继续加载上下文。我们可以以编程方式调用admin的initialize()方法再次尝试。如果我们希望将此条件视为致命，请将admin的fatalIfBrokerNotAvailable属性设置为true。
这样的话，上下文就会初始化失败。

	@Autowired
	private KafkaAdmin admin;
	
	public void testMethod() {
		admin.setFatalIfBrokerNotAvailable(true);
	}

如果broker(代理)支持它（1.0.0或更高版本），则如果发现现有主题的分区数少于分区，则admin会增加分区数NewTopic.numPartitions。

有关更高级的功能（例如将分区分配给副本），可以直接使用AdminClient。以下示例显示了如何执行此操作：

@Autowired
private KafkaAdmin admin;

...

    AdminClient client = AdminClient.create(admin.getConfig());
    ...
    client.close();

1.1.2 发送消息

这部分讲解如何发送消息。

1.1.2.1 使用KafkaTemplate

本节介绍如何使用KafkaTemplate发送消息。

概览

KafkaTemplate包装生产者并提供方便的方法来将数据发送到Kafka主题。以下清单显示了KafkaTemplate的相关方法：

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

}

sendDefault API要求已为template提供默认topic。

API将时间戳作为参数，并将此时间戳存储在记录中。如何存储用户提供的时间戳取决于Kafka topic上配置的时间戳类型。如果主题配置为使用CREATE_TIME，则记录用户指定的时间戳（如果未指定，则生成）。如果主题配置为使用LOG_APPEND_TIME，则会忽略用户指定的时间戳，并且代理会在本地代理时间中添加。

metrics和partitions 方法委托给底层Producer上的相同方法。 execute方法提供对底层Producer的直接访问。

要使用该模板，您可以配置生产者工厂并在模板的构造函数中提供它。以下示例显示了如何执行此操作：

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

您还可以使用标准<bean/>定义来配置模板。

然后，要使用该模板，您可以调用其中一个方法。

将方法与Message<?> 参数一起使用时，topic，partition(分区)和key(密钥)信息将在包含以下各项的header中提供：

KafkaHeaders.TOPIC
KafkaHeaders.PARTITION_ID
KafkaHeaders.MESSAGE_KEY
KafkaHeaders.TIMESTAMP

消息payload是数据。

（可选）您可以使用ProducerListener配置KafkaTemplate，以获取包含send（成功或失败）结果的异步回调，而不是等待Future完成。以下清单显示了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();

}

默认情况下，模板会配置一个LoggingProducerListener，它记录错误，并在发送成功时不执行任何操作。

仅当isInterestedInSuccess返回true时才调用onSuccess。

为方便起见，提供了抽象的ProducerListenerAdapter，以便于我们只想实现其中一个方法。它为isInterestedInSuccess返回false。

请注意，send方法返回一个ListenableFuture<SendResult>。您可以向监听器注册回调，以异步方式接收发送结果。以下考试说明了如何执行此操作：

ListenableFuture<SendResult<Integer, String>> future = template.send("something");
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {

    @Override
    public void onSuccess(SendResult<Integer, String> result) {
        ...
    }

    @Override
    public void onFailure(Throwable ex) {
        ...
    }

});

SendResult有两个属性，一个是ProducerRecord和RecordMetadata。有关这些对象的信息，请参阅Kafka API文档。

如果您希望阻止发送线程等待结果，您可以调用future的get()方法。您可能希望在等待之前调用flush()，或者为方便起见，模板具有一个带有autoFlush参数的构造函数，该参数会导致模板在每次发送时执行flush()方法。但请注意，刷新可能会显着降低性能。

例子

本节显示向Kafka发送消息的示例：

无阻塞异步

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

1.1.2.2 事务

本节描述Spring for Apache Kafka如何支持事务。

概览

0.11.0.0客户端库添加了对事务的支持。 Apache Kafka的Spring通过以下方式增加了支持：

KafkaTransactionManager：与普通的Spring事务支持一起使用（@
Transaction，TransactionTemplate等）。
KafkaMessageListenerContainer事务性
与KafkaTemplate的本地事务

通过向DefaultKafkaProducerFactory提供transactionIdPrefix来启用事务。在这种情况下，工厂不是管理单个共享生产者，而是维护事务生成器的缓存。当用户在生产者上调用close()时，它将返回到缓存以供重用，而不是实际关闭。每个生成器的transactional.id属性是transactionIdPrefix + n，其中n以0开头并为每个新生成器递增，除非事务由具有基于记录的侦听器的侦听器容器启动。在这种情况下，transactional.id是<transactionIdPrefix>.<group.id>.<topic>.<partition>。这是为了正确支持fencing zombies，如此处所述。在1.3.7,2.0.6,2.1.10和2.2.0版本中添加了此新行为。如果您希望恢复到以前的行为，可以将DefaultKafkaProducerFactory上的producerPerConsumerPartition属性设置为false。

虽然批处理监听器支持事务，但不支持zombie fencing ，因为批处理可能包含来自多个topic或partitions(分区)的记录。

使用 KafkaTransactionManager

KafkaTransactionManager是Spring Framework的PlatformTransactionManager的实现。它在构造函数中提供了对生产者工厂的引用。如果您提供自定义生产者工厂，它必须支持事务。请参见ProducerFactory.transactionCapable()。

您可以使用KafkaTransactionManager和普通的Spring事务支持（@Transactional，TransactionTemplate等）。如果事务处于活动状态，则在事务范围内执行的任何KafkaTemplate操作都使用事务的Producer。这个管理器会根据成功或失败提交或回滚交易。您必须将KafkaTemplate配置为使用与事务管理器相同的ProducerFactory。

事务监听器容器和完全一次处理

我们可以为监听器容器提供KafkaAwareTransactionManager实例。如此配置，容器在调用监听器之前启动事务。监听器执行的任何KafkaTemplate操作都参与事务。如果监听器成功处理记录（或多个记录，当使用BatchMessageListener时），则在事务管理器提交事务之前，容器通过使用producer.sendOffsetsToTransaction() ）将offsets(偏移)发送到事务。如果监听器抛出异常，则回滚事务并重新定位使用者，以便在下次轮询时检索回滚记录。有关详细信息和处理重复失败的记录，请参阅After-rollback Processor。

事务同步

如果需要将Kafka事务与某个其他事务同步，请使用适当的事务管理器（支持同步的事务管理器，例如DataSourceTransactionManager）配置监听器容器。从监听器对事务KafkaTemplate执行的任何操作都参与单个事务。在控制事务之后立即提交（或回滚）Kafka事务。在退出监听器之前，您应该调用模板的sendOffsetsToTransaction方法之一（除非您使用ChainedKafkaTransactionManager）。为方便起见，监听器容器将其使用者组ID绑定到线程，因此，通常，您可以使用第一种方法。以下清单显示了两种方法签名：

void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets);

void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId);

以下示例显示如何使用sendOffsetsToTransaction方法的第一个签名：

@Bean
KafkaMessageListenerContainer container(ConsumerFactory<String, String> cf,
            final KafkaTemplate template) {
    ContainerProperties props = new ContainerProperties("foo");
    props.setGroupId("group");
    props.setTransactionManager(new SomeOtherTransactionManager());
    ...
    props.setMessageListener((MessageListener<String, String>) m -> {
        template.send("foo", "bar");
        template.send("baz", "qux");
        template.sendOffsetsToTransaction(
            Collections.singletonMap(new TopicPartition(m.topic(), m.partition()),
                new OffsetAndMetadata(m.offset() + 1)));
    });
    return new KafkaMessageListenerContainer<>(cf, props);
}

要提交的offset(偏移量)大于监听器处理的记录的offset(偏移量)。

只有在使用事务同步时才应该调用它。当监听器容器配置为使用KafkaTransactionManager时，它负责将offset(偏移量)发送到事务。

使用 ChainedKafkaTransactionManager

ChainedKafkaTransactionManager在2.1.3版中引入。这是ChainedTransactionManager的子类，可以只有一个KafkaTransactionManager。由于它是KafkaAwareTransactionManager，容器可以使用与使用简单的KafkaTransactionManager配置容器时相同的方式将offset（偏移）发送到事务。这提供了另一种同步事务的机制，而不必将offset（偏移量）发送到监听器代码中的事务。您应该按所需顺序链接事务管理器，并在ContainerProperties中提供ChainedTransactionManager。

KafkaTemplate本地事务

您可以使用KafkaTemplate在本地事务中执行一系列操作。以下示例显示了如何执行此操作：

boolean result = template.executeInTransaction(t -> {
    t.sendDefault("thing1", "thing2");
    t.sendDefault("cat", "hat");
    return true;
});

回调中的参数是模板本身（this）。如果回调正常退出，则提交事务。如果抛出异常，则回滚事务。

如果正在处理KafkaTransactionManager（或同步）事务，则不使用它。相反，使用新的“nested（嵌套）”事务。

使用ReplyingKafkaTemplate

2.1.3版引入了KafkaTemplate的子类来提供 request/reply语义。该类名为ReplyingKafkaTemplate，并且有一个方法（除了超类中的方法）。以下清单显示了该方法的签名：

RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record);

返回值是一个ListenableFuture，它是用返回值异步填充的（或者是超时的异常）。返回值还有一个sendFuture属性，它是调用KafkaTemplate.send()的返回值。您可以使用此未来来确定发送操作的结果。

以下Spring Boot应用程序显示了如何使用该功能的示例：

@SpringBootApplication
public class KRequestingApplication {

    public static void main(String[] args) {
        SpringApplication.run(KRequestingApplication.class, args).close();
    }

    @Bean
    public ApplicationRunner runner(ReplyingKafkaTemplate<String, String, String> template) {
        return args -> {
            ProducerRecord<String, String> record = new ProducerRecord<>("kRequests", "foo");
            RequestReplyFuture<String, String, String> replyFuture = template.sendAndReceive(record);
            SendResult<String, String> sendResult = replyFuture.getSendFuture().get();
            System.out.println("Sent ok: " + sendResult.getRecordMetadata());
            ConsumerRecord<String, String> consumerRecord = replyFuture.get();
            System.out.println("Return value: " + consumerRecord.value());
        };
    }

    @Bean
    public ReplyingKafkaTemplate<String, String, String> replyingTemplate(
            ProducerFactory<String, String> pf,
            ConcurrentMessageListenerContainer<Long, String> repliesContainer) {

        return new ReplyingKafkaTemplate<>(pf, repliesContainer);
    }

    @Bean
    public ConcurrentMessageListenerContainer<String, String> repliesContainer(
            ConcurrentKafkaListenerContainerFactory<String, String> containerFactory) {

        ConcurrentMessageListenerContainer<String, String> repliesContainer =
                containerFactory.createContainer("replies");
        repliesContainer.getContainerProperties().setGroupId("repliesGroup");
        repliesContainer.setAutoStartup(false);
        return repliesContainer;
    }

    @Bean
    public NewTopic kRequests() {
        return new NewTopic("kRequests", 10, (short) 2);
    }

    @Bean
    public NewTopic kReplies() {
        return new NewTopic("kReplies", 10, (short) 2);
    }

}

请注意，我们可以使用Spring Boot的自动配置容器工厂来创建reply(回复)容器。

该模板设置了一个名为KafkaHeaders.CORRELATION_ID的标头，该标头必须由服务器端回送。

在这种情况下，以下@KafkaListener应用程序响应：

@SpringBootApplication
public class KReplyingApplication {

    public static void main(String[] args) {
        SpringApplication.run(KReplyingApplication.class, args);
    }

    @KafkaListener(id="server", topics = "kRequests")
    @SendTo // use default replyTo expression
    public String listen(String in) {
        System.out.println("Server received: " + in);
        return in.toUpperCase();
    }

    @Bean
    public NewTopic kRequests() {
        return new NewTopic("kRequests", 10, (short) 2);
    }

    @Bean // not required if Jackson is on the classpath
    public MessagingMessageConverter simpleMapperConverter() {
        MessagingMessageConverter messagingMessageConverter = new MessagingMessageConverter();
        messagingMessageConverter.setHeaderMapper(new SimpleKafkaHeaderMapper());
        return messagingMessageConverter;
    }

}

@KafkaListener基础结构回显相关ID并确定回复topic(主题)。

有关发送回复的详细信息，请参阅使用@SendTo转发侦听器结果。模板使用默认header KafKaHeaders.REPLY_TOPIC来指示回复所针对的主题。

从2.2版开始，模板尝试从配置的回复容器中检测回复topic(主题)或partition(分区)。如果容器配置为侦听单个主题或单个TopicPartitionInitialOffset，则它用于设置回复header。如果以其他方式配置容器，则用户必须设置回复header。在这种情况下，在初始化期间写入INFO日志消息。以下示例使用KafkaHeaders.REPLY_TOPIC

record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, "kReplies".getBytes()));

使用单个回复TopicPartitionInitialOffset进行配置时，只要每个实例监听不同的分区，就可以对多个模板使用相同的回复主题。使用单个回复主题进行配置时，每个实例必须使用不同的group.id。在这种情况下，所有实例都会收到每个回复，但只有发送请求的实例才会找到相关ID。这可能对自动扩展很有用，但会增加额外网络流量的开销，并且丢弃每个不需要的回复的成本很低。使用此设置时，我们建议您将模板的sharedReplyTopic设置为true，这会降低对DEBUG的意外答复的日志记录级别，而不是默认的ERROR。

如果您有多个客户端实例，并且未按照前一段中的讨论进行配置，则每个实例都需要一个专用的回复主题。
另一种方法是设置KafkaHeaders.REPLY_PARTITION并为每个实例使用专用分区。
Header包含一个四字节的int（big-endian）。
服务器必须使用此标头将回复路由到正确的主题（@KafkaListener执行此操作）。
但是，在这种情况下，回复容器不能使用Kafka的组管理功能，必须配置为侦听固定分区（通过在其ContainerProperties构造函数中使用TopicPartitionInitialOffset）。
DefaultKafkaHeaderMapper要求Jackson在类路径上（对于@KafkaListener）。
如果它不可用，则消息转换器没有标头映射器，因此您必须使用SimpleKafkaHeaderMapper配置MessagingMessageConverter，如前所示。

本节完~

本文标题：第二集 Spring for Apache Kafka 配置主题和发送消息

本文链接：https://blog.quwenai.cn/post/3005.html

作者授权：除特别说明外，本文由 toobug 原创编译并授权 TOOBUG信息安全网刊载发布。

本文链接：https://blog.quwenai.cn/post/3005.html

分享到：

正文

第二集 Spring for Apache Kafka 配置主题和发送消息

1. 参考

1.1 使用Spring for Apache Kafka

1.1.1 配置Topic

1.1.2 发送消息

1.1.2.1 使用KafkaTemplate

概览

例子

1.1.2.2 事务

概览

使用 KafkaTransactionManager

事务监听器容器和完全一次处理

事务同步

使用 ChainedKafkaTransactionManager

KafkaTemplate本地事务

使用ReplyingKafkaTemplate

相关阅读

第一集 Spring for Apache Kafka入门

0x05 Spring Boot 2.x之日志系统详解

2.2 Spring Framework 5.x 之 Container（容器）概览

0x04 Berkeley DB XMl Java版之添加 XML Document到Containers

发表评论取消回复

还没有评论，来说两句吧...

目录[+]