247 lines
12 KiB
Markdown
247 lines
12 KiB
Markdown
# Spring Kafka
|
||
## 连接到kafka
|
||
### 运行时切换bootstrap servers
|
||
从2.5版本开始，KafkaAdmin、ProducerFactory、ConsumerFactory都继承于`KafkaResourceFactory`抽象类。通过调用`KafkaResourceFactory`抽象类的`setBootstrapServersSupplier(Supplier<String> bootstrapServersSupplier)`方法，可以在运行时动态的切换bootstrap servers。该Supplier将会在新建连接获取bootstrap servers时被调用。
|
||
|
||
> #### 切换bootstrap后关闭旧consumer和producer
|
||
> kafka consumer和producer通常都是基于长连接的，在调用setBootstrapServersSupplier在运行时切换bootstrap servers后，如果想要关闭现存的producer，可以调用`DefaultKafkaProducerFactory`的`reset`方法。如果想要关闭现存的consumer，可以调用`KafkaListenerEndpointRegistry`的`close`方法（调用close后再调用start），或是调用其他listener container的close和start方法。
|
||
|
||
#### ABSwitchCluster
|
||
为了方便起见，framework提供了`ABSwitchCluster`类，该类支持两套bootstrap servers集合，在任一时刻，只有其中一套bootstrap servers起作用。ABSwitchCluster类继承Supplier\<String\>接口，将`ABSwitchCluster`对象提供给consumer factory, producer factory, KafkaAdmin后，如果想要切换bootstrap servers，可以调用ABSwitchCluster类的`primary`和`secondary`方法，并关闭生产者和消费者的旧实例（关闭生产者旧实例，在producer factory上调用reset方法，用于创建到新bootstrap servers的连接；对于消费者实例，可以对所有listener container先调用close方法再调用start方法，当使用@KafkaListener注解时，需要对`KafkaListenerEndpointRegistry`bean对象调用close和start方法。
|
||
|
||
### Factory Listener
|
||
从2.5版本开始，`DefaultKafkaProducerFactory`和`DefaultKafkaConsumerFactory`都可以配置Listener，通过配置Listener可以监听生产者或消费者实例的创建和关闭。
|
||
|
||
```java
|
||
// producer listener
|
||
interface Listener<K, V> {
|
||
|
||
default void producerAdded(String id, Producer<K, V> producer) {
|
||
}
|
||
|
||
default void producerRemoved(String id, Producer<K, V> producer) {
|
||
}
|
||
|
||
}
|
||
```
|
||
```java
|
||
// consumer listener
|
||
interface Listener<K, V> {
|
||
|
||
default void consumerAdded(String id, Consumer<K, V> consumer) {
|
||
}
|
||
|
||
default void consumerRemoved(String id, Consumer<K, V> consumer) {
|
||
}
|
||
|
||
}
|
||
```
|
||
再上述接口中，id代表再factory bean对象名称后追加client-id属性，二者通过`.`分隔。
|
||
|
||
## 配置Topic
|
||
如果在当前应用上下文中定义了KafkaAdmin bean对象，kafkaAdmin可以自动的添加topic到broker。为了实现topic的自动添加，可以定义一个`NewTopic`类型的bean对象，kafkaAdmin会自动将该topic添加到broker中。
|
||
|
||
为了方便topic的创建，2.3版本中引入了TopicBuilder类。
|
||
```java
|
||
@Bean
|
||
public KafkaAdmin admin() {
|
||
Map<String, Object> configs = new HashMap<>();
|
||
configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
|
||
return new KafkaAdmin(configs);
|
||
}
|
||
|
||
@Bean
|
||
public NewTopic topic1() {
|
||
return TopicBuilder.name("thing1")
|
||
.partitions(10)
|
||
.replicas(3)
|
||
.compact()
|
||
.build();
|
||
}
|
||
|
||
@Bean
|
||
public NewTopic topic2() {
|
||
return TopicBuilder.name("thing2")
|
||
.partitions(10)
|
||
.replicas(3)
|
||
.config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
|
||
.build();
|
||
}
|
||
|
||
@Bean
|
||
public NewTopic topic3() {
|
||
return TopicBuilder.name("thing3")
|
||
.assignReplicas(0, List.of(0, 1))
|
||
.assignReplicas(1, List.of(1, 2))
|
||
.assignReplicas(2, List.of(2, 0))
|
||
.config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
|
||
.build();
|
||
}
|
||
```
|
||
|
||
从2.6版本开始，创建NewTopic时可以省略partitions()和replicas()方法的调用，此时创建的topic将会使用broker中默认的配置。支持该特性要求broker版本至少为2.4.0。
|
||
|
||
```java
|
||
@Bean
|
||
public NewTopic topic4() {
|
||
return TopicBuilder.name("defaultBoth")
|
||
.build();
|
||
}
|
||
|
||
@Bean
|
||
public NewTopic topic5() {
|
||
return TopicBuilder.name("defaultPart")
|
||
.replicas(1)
|
||
.build();
|
||
}
|
||
|
||
@Bean
|
||
public NewTopic topic6() {
|
||
return TopicBuilder.name("defaultRepl")
|
||
.partitions(3)
|
||
.build();
|
||
}
|
||
```
|
||
从版本2.7开始，可以在`KafkaAdmin.NewTopics`的bean对象中声明多个NewTopic对象：
|
||
```java
|
||
@Bean
|
||
public KafkaAdmin.NewTopics topics456() {
|
||
return new NewTopics(
|
||
TopicBuilder.name("defaultBoth")
|
||
.build(),
|
||
TopicBuilder.name("defaultPart")
|
||
.replicas(1)
|
||
.build(),
|
||
TopicBuilder.name("defaultRepl")
|
||
.partitions(3)
|
||
.build());
|
||
}
|
||
```
|
||
> 当使用spring boot时，KafkaAdmin对象将会被自动注册，故而只需要定义NewTopic bean对象即可。
|
||
|
||
默认情况下，如果kafka broker不可用，会输出日志进行记录，但是此时context的载入还会继续，后续可以手动调用KafkaAdmin的`initalize`方法和进行重试。如果想要在kafka broker不可用时，停止context的载入，可以将kafka Admin`fatalIfBrokerNotAvailable`属性设置为true，此时context会初始化失败。
|
||
|
||
从版本2.7开始，KafkaAdmin提供了两个方法用于在运行时动态创建和检测Topic：
|
||
- `createOrModifyTopics`
|
||
- `describeTopics`
|
||
|
||
从版本2.9.10、3.0.9开始，KafkaAdmin提供了`setCreateOrModifyTopic(Predicate<org.apache.kafka.clients.admin.NewTopic> createOrModifyTopic)`接口，该接口接收一个Predicate\<NewTopic\>参数，通过该predicate可以判断是否一个NewTopic bean应该被该kafkaAdmin创建或修改。该方法通常用于上下文中含有多个KafkaAdmin bena对象，每个kafkaAdmin对应不同的broker集群，在上下文中含有多个NewTopic对象时，可以通过predicate判断每个topic应该属性哪个amdin。
|
||
|
||
## 发送消息
|
||
KafkaTemplate类对KafkaProducer进行了包装，提供了如下接口用于向kafka topic发送消息。
|
||
```java
|
||
CompletableFuture<SendResult<K, V>> sendDefault(V data);
|
||
|
||
CompletableFuture<SendResult<K, V>> sendDefault(K key, V data);
|
||
|
||
CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
|
||
|
||
CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
|
||
|
||
CompletableFuture<SendResult<K, V>> send(String topic, V data);
|
||
|
||
CompletableFuture<SendResult<K, V>> send(String topic, K key, V data);
|
||
|
||
CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
|
||
|
||
CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
|
||
|
||
CompletableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
|
||
|
||
CompletableFuture<SendResult<K, V>> send(Message<?> message);
|
||
|
||
Map<MetricName, ? extends Metric> metrics();
|
||
|
||
List<PartitionInfo> partitionsFor(String topic);
|
||
|
||
<T> T execute(ProducerCallback<K, V, T> callback);
|
||
|
||
<T> T executeInTransaction(OperationsCallback<K, V, T> callback);
|
||
|
||
// Flush the producer.
|
||
void flush();
|
||
|
||
interface ProducerCallback<K, V, T> {
|
||
|
||
T doInKafka(Producer<K, V> producer);
|
||
|
||
}
|
||
|
||
interface OperationsCallback<K, V, T> {
|
||
|
||
T doInOperations(KafkaOperations<K, V> operations);
|
||
|
||
}
|
||
```
|
||
其中，sendDefault接口需要向KafkaTemplate提供一个默认的topic。
|
||
|
||
kafkaTemplate中部分api接收timestamp作为参数，并且将timestamp存储到record中。接口中指定的timestamp参数如何存储，取决于kafka topic中配置的timestamp类型。如果topic中timestamp类型被配置为`CREATE_TIME`，那么用户指定的timestamp参数将会被使用（如果用户没有指定timestamp，那么会自动创建timestamp，producer会在发送时将timestamp指定为System.currentTimeMillis()）。如果topic中timstamp类型被配置为`LOG_APPEND_TIME`，那么用户指定的timestamp将会被丢弃，而broker则会负责为timestamp赋值。
|
||
|
||
mertics和partitions方法则会被委派给了底层KafkaProducer的同名方法，execute接口则是提供了对底层KafkaProducer的直接访问。
|
||
|
||
要使用KafkaTemplate，可以配置一个producer factory并将其提供给KafkaTemplate的构造方法。如下展示了如何配置一个KafkaTemplate：
|
||
```java
|
||
@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, IntegerSerializer.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());
|
||
}
|
||
```
|
||
从2.5开始，创建KafkaTemplate时可以基于factory进行创建，但是覆盖factory中的配置属性，具体示例如下：
|
||
```java
|
||
@Bean
|
||
public KafkaTemplate<String, String> stringTemplate(ProducerFactory<String, String> pf) {
|
||
return new KafkaTemplate<>(pf);
|
||
}
|
||
|
||
@Bean
|
||
public KafkaTemplate<String, byte[]> bytesTemplate(ProducerFactory<String, byte[]> pf) {
|
||
return new KafkaTemplate<>(pf,
|
||
Collections.singletonMap(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class));
|
||
}
|
||
```
|
||
|
||
当使用KafkaTemplate接收`Message\<?\>`类型的参数时，可以将topic、partition、key和timestamp参数指定在Message的header中，header中包含如下条目：
|
||
- KafkaHeaders.TOPIC
|
||
- KafkaHeaders.PARTITION
|
||
- KafkaHeaders.KEY
|
||
- KafkaHeaders.TIMESTAMP
|
||
|
||
除了调用发送方法获取CompletableFuture外，还可以为KafkaTemplate配置一个ProducerListener，从而在消息发送完成（成功或失败）后执行一个异步的回调。如下是ProducerListener接口的定义：
|
||
```java
|
||
public interface ProducerListener<K, V> {
|
||
|
||
void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata);
|
||
|
||
void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata,
|
||
Exception exception);
|
||
|
||
}
|
||
```
|
||
默认情况下，KafkaTemplate配置了一个LoggingProducerListener，会在发送失败时打印失败日志，在发送成功时并不做任何事。并且为了方便起见，方法的默认实现已经被提供，可以只覆盖其中一个方法。
|
||
|
||
send方法默认返回的是CompletableFuture类型，可以在发送完成之后为future注册一个回调：
|
||
```java
|
||
CompletableFuture<SendResult<Integer, String>> future = template.send("myTopic", "something");
|
||
future.whenComplete((result, ex) -> {
|
||
...
|
||
});
|
||
```
|
||
其中，Throwable类型的ex可以被转化为`KafkaProducerException`，该类型的failedProducerRecord属性可以获取发送失败的record。
|
||
|
||
如果想要同步调用KafkaTemplate的发送方法并且等待返回结果，可以调用返回值CompletableFuture类型的get方法来同步等待。通常情况下，调用`CompletableFuture.get`时，推荐使用带超时参数的方法。如果在Producer配置中指定了`linger.ms`，那么在等待返回结果之前需要调用KafkaTemplate的flush方法。为了方便，KafkaTemplate提供了带autoFlush参数的构造器版本，如果设置autoFlush为true，kafkaTemplate在每次发送消息时都会调用flush方法。
|