SpringCloud 用法示例

此示例說明了如何在用戶應(yīng)用程序中手動確認(rèn)偏移。

本示例要求將spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOffset設(shè)置為false。在您的示例中使用相應(yīng)的輸入通道名稱。

@SpringBootApplication
@EnableBinding(Sink.class)
public class ManuallyAcknowdledgingConsumer {

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

 @StreamListener(Sink.INPUT)
 public void process(Message<?> message) {
     Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
     if (acknowledgment != null) {
         System.out.println("Acknowledgment provided");
         acknowledgment.acknowledge();
     }
 }
}

Apache Kafka 0.9支持客戶端和代理之間的安全連接。要利用此功能，請遵循Apache Kafka文檔中的準(zhǔn)則以及Confluent文檔中的Kafka 0.9 安全準(zhǔn)則。使用spring.cloud.stream.kafka.binder.configuration選項為活頁夾創(chuàng)建的所有客戶端設(shè)置安全性屬性。

例如，要將security.protocol設(shè)置為SASL_SSL，請設(shè)置以下屬性：

spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL

可以以類似方式設(shè)置所有其他安全屬性。

使用Kerberos時，請遵循參考文檔中的說明來創(chuàng)建和引用JAAS配置。

Spring Cloud Stream支持通過使用JAAS配置文件并使用Spring Boot屬性將JAAS配置信息傳遞到應(yīng)用程序。

 java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \
   --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \
   --spring.cloud.stream.bindings.input.destination=stream.ticktock \
   --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT

 java --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \
   --spring.cloud.stream.bindings.input.destination=stream.ticktock \
   --spring.cloud.stream.kafka.binder.autoCreateTopics=false \
   --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT \
   --spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true \
   --spring.cloud.stream.kafka.binder.jaas.options.storeKey=true \
   --spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab \
   --spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM

KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka_client.keytab"
    principal="kafka-client-1@EXAMPLE.COM";
};

如果希望暫停使用但不引起分區(qū)重新平衡，則可以暫停并恢復(fù)使用方。通過將Consumer作為參數(shù)添加到@StreamListener中，可以簡化此操作。要恢復(fù)，需要為ListenerContainerIdleEvent實例使用ApplicationListener。事件的發(fā)布頻率由idleEventInterval屬性控制。由于使用者不是線程安全的，因此必須在調(diào)用線程上調(diào)用這些方法。

以下簡單的應(yīng)用程序顯示了如何暫停和恢復(fù)：

@SpringBootApplication
@EnableBinding(Sink.class)
public class Application {

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

	@StreamListener(Sink.INPUT)
	public void in(String in, @Header(KafkaHeaders.CONSUMER) Consumer<?, ?> consumer) {
		System.out.println(in);
		consumer.pause(Collections.singleton(new TopicPartition("myTopic", 0)));
	}

	@Bean
	public ApplicationListener<ListenerContainerIdleEvent> idleListener() {
		return event -> {
			System.out.println(event);
			if (event.getConsumer().paused().size() > 0) {
				event.getConsumer().resume(event.getConsumer().paused());
			}
		};
	}

}