Spring boot 项目中如何进行kafka Stream app 开发

目录

• Kafka Stream
• Kafka Stream的特点
• 1. 导入依赖
• 2. 示例代码(伪代码)
• 3. 一些注意事项和说明
• 4. kafka Stream 的一些方法说明

Kafka Stream

Kafka Stream是Apache Kafka从0.10版本引入的一个新Feature。它是提供了对存储于Kafka内的数据进行流式处理和分析的功能。

Kafka Stream的特点

• Kafka Stream提供了一个非常简单而轻量的Library，它可以非常方便地嵌入任意Java应用中，也可以任意方式打包和部署
• 除了Kafka外，无任何外部依赖
• 充分利用Kafka分区机制实现水平扩展和顺序性保证
• 通过可容错的state store实现高效的状态操作（如windowed join和aggregation）
• 支持正好一次处理语义
• 提供记录级的处理能力，从而实现毫秒级的低延迟
• 支持基于事件时间的窗口操作，并且可处理晚到的数据（late arrival of records）
• 同时提供底层的处理原语Processor（类似于Storm的spout和bolt），以及高层抽象的DSL（类似于Spark的map/group/reduce）

下面介绍Spring boot 项目中进行kafka Stream app 开发的详细过程。

1. 导入依赖

      <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>3.6.2</version>
      </dependency>

2. 示例代码(伪代码)

这段伪代码只是为了举例设置的场景，业务场景并不一定合适

@Slf4j
@Component
public class MyKafkaStreamProcessor {
	@Value("${spring.kafka.bootstrap-servers}")
	private String bootstrapServers;
	@PostConstruct
	private void init () {
		String appId = "my-kafka-streams-app";
		myKafkaStreams(appId);
		log.info("✅ Kafka Streams:{} 初始化完成，开始监听 topic: {}", appId, "source-topic");
	}
	public void myKafkaStreams(String appId) {
		/*=======配置=======*/
		Properties config  = new Properties();
		config.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
		config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
		StreamsBuilder builder = new StreamsBuilder();
		/*=======构建拓扑结构=======*/
		// 数据清洗
		KStream<String, Order> stream = builder
			.stream("source-topic", Consumed.with(Serdes.String(), new jsonSerde<>(Order.class)))
			.mapValues(value -> {
				// do something
				return value;
			});
		// 过滤出从app创建的订单 并进行处理
		stream.filter((k, v) -> Order.getSource.equals("app"))
			.foreach((k, v) -> {
			编程客栈	// do something
			});
		// 发送到第一个topic
		stream.mapValu编程es(value -> JSON.toJSONString(value), Named.as("to-the-first-target-topic-processor"))
			.to("the-first-target-topic",cYaMGFDKYj Produced.with(Serdes.String(), Serdes.String()));
		// 发送到第二个topic
		stream.filter((k, v) -> {
				// filter something
			})
			.mapValues(value -> {
				// map to another object
			}, Named.as("to-the-second-target-topic-processor"))
			.to("the-second-target-topic", Produced.with(Serdes.String(), Serdes.String()));
		/*=======创建KafkaStreams=======*/
		KafkaStreams streams = new KafkaStreams(builder.build(), config);
		/*=======设置异常处理器=======*/
		streams.setUncaughtExceptionHandler(new CustomStreamsUncaughtExceptionHandler());
		/*=======启动streams=======*/
		streams.start(js);
		/*=======添加jvm hook 确保streams安全退出=======*/
		Runtime.getRuntime().addShutdownHook(new Thread(() -> {
			log.info("关闭 Kafka Streams:{}...", appId);
			streams.close();
			log.info("Kafka Streams:{}已经关闭！", appId);
		}));
	}
}

@Slf4j
public class CustomStreamsUncaughtExceptionHandler implements StreamsUncaughtExceptionHandler {
	/**
	 * Inspect the exception received in a stream thread and respond with an action.
	 *
	 */
	@Override
	public StreamThreadExceptionResponse handle(Throwable throwable) {
		log.error("Kafka Streams 线程发生未捕获异常: {}", ExceptionUtil.stacktraceToString(throwable));
		// 选择处理策略（以下三选一）：
		// 1. 替换线程（继续运行）
		return StreamThreadExceptionResponse.REPLACE_THREAD;
		// 2. 关闭整个 Streams 应用
		// return StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
		// 3. 关闭整个 JVM
		// return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
	}
}

3. 一些注意事项和说明

• kafka stream 的处理部分集中在构建的拓扑中，其他部分大同小异
• 在配置部分 StreamsConfig.APPLICATION_ID_CONFIG 这个参数是必须的，且不能重复，否则会启动失败，StreamsConfig.BOOTSTRAP_SERVERS_CONFIG 是kafka的IP与端口
• 需要注意的是kafka的 KStream 与 java 中的 stream并不相同，在java中 stream只能被消费一次，但是kstream 可以被消费多次，在上面的demo中可以看到，同一个 kstream 被多次消费，且kstream中的数据是不可变的，也就是无论在上一个处理器（processor）对数据进行了何种处理，下一个处理器从kstream 中获取的数据依旧是原来的数据
• 在kafka stream app中应该对可能会抛出的异常进行处理，而不是全部交给UncaughtExceptionHandler，UncaughtExceptionHandler应该只处理哪些无法预料的异常
• 如果kafka stream app 捕获未处理异常之后的处理策略也是替换线程，那么kafka stream app 中如果抛出未捕获异常，那么这个消费者组就会进入再平衡状态（PreparingRebalance），老的消费者从消费者组中剔除，新的消费者加入消费者组，然后再开始消费，注意这种替换线程的处理策略可能导致消息重放，也就是原本的线程消费的offset没有提交导致新的线程会python重复消费之前已经被消费的数据，如果业务会因为消息重放出现异常，建议做幂等

4. kafka Stream 的一些方法说明

• stream()
  • stream()方法是从源topic获取数据的方法，示例中第一个参数是字符串，也就是源topic的名称，
  • 第二个参数是 Consumed ，用来定义对于消息的key与value反序列化的规则，
  • 示例中将key序列化为string， value 序列化为order对象，需要注意的是，如果在配置的config中没有设置适用于整个stream app的序列化与反序列化规则，那么后续的 to()中必须要指定序列化规则
• filter()
  • filter()的用法与java stream 中的filter()一致，这里不做说明
• mapValues()
  • mapValues()的用法，是只对消息的value进行操作，比如将value转换为其他对象。这个方法不会对key进行修改
• map()
  • 与mapValues()类似，但是可以修改消息的key
• foreach()
  • 与java stream 的 foreach()类似，也是一个终结方法
• to()
  • 终结方法，用于将数据发送到另外的topic，示例中第一个参数是目标topic， 第二个参数是Produced，用于定义key和value的序列化规则
  • 除了stream()和to()之外，其他方法基本都可以传一个参数 Named，这个参数是为每个处理器节点命名，如果不传则自动生成，但是在同一个kstream中 命名不能重复。这个名称不会影响功能，但是如果有一个名称可以在后续调试和监控中提供一点帮助

到此这篇关于Spring boot 项目中如何进行kafka Stream app 开发的文章就介绍到这了,更多相关Spring boot kafka Stream app 开发内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！