springboot~kafka-stream实现实时统计 - ToB企服应用市场:ToB评测及商务社交产业平台

[Source Processor] -> [Processor Topology] -> [Sink Processor]

复制代码

<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.5.5.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.5.1</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>2.5.1</version>
</dependency>

复制代码

@Configuration
@EnableKafkaStreams
public class KafkaStreamConfig {
private static final int MAX_MESSAGE_SIZE = 16 * 1024 * 1024;
@Value("${spring.kafka.bootstrap-servers}")
private String hosts;
@Value("${spring.kafka.consumer.group-id}")
private String group;
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
Map<String, Object> props = new HashMap<>();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
props.put(StreamsConfig.APPLICATION_ID_CONFIG, group + "_stream_aid");
props.put(StreamsConfig.CLIENT_ID_CONFIG, group + "_stream_cid");
props.put(StreamsConfig.RETRIES_CONFIG, 3);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");//从最近的消息开始消费
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
return new KafkaStreamsConfiguration(props);
}
}

复制代码

@Configuration
@Slf4j
public class KafkaStreamListener {
@Autowired
ReportLoginTypeMapper reportLoginTypeMapper;
@KafkaListener(topics = "total_record")
public void listen(ConsumerRecord<String, String> record) {
// 将时间戳转换为本地日期时间
LocalDateTime dateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(record.timestamp()), ZoneId.systemDefault());
ReportLoginType reportLoginType=new ReportLoginType();
reportLoginType.setLoginType(record.key());
reportLoginType.setCreateAt(dateTime);
reportLoginType.setCount(Integer.parseInt(record.value()));
reportLoginTypeMapper.insert(reportLoginType);
}
@Bean
public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
KStream<String, String> stream = streamsBuilder.stream("KC_LOGIN");
KStream<String, String> serializedStream = stream.mapValues(jsonString -> {
// 分组依据
if (JSONUtil.parseObj(jsonString).containsKey("details")) {
JSONObject details = JSONUtil.parseObj(jsonString).getJSONObject("details");
if (details.containsKey("loginType")) {
String loginType = details.getStr("loginType");
return loginType;
}
return "";
}
else {
return "";
}
});
/**
* 处理消息的value
*/
serializedStream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
@Override
public Iterable<String> apply(String value) {
return Arrays.asList(value.split(" "));
}
}).filter((key, value) -> !value.equals(""))
// 按照value进行聚合处理
.groupBy((key, value) -> value)// 这进而的value是kafka的消息内容
// 时间窗口
.windowedBy(TimeWindows.of(Duration.ofSeconds(60)))
// 统计单词的个数
.count()
// 转换为kStream
.toStream().map((key, value) -> {
// key是分组的key,value是分组count的结果
return new KeyValue<>(key.key().toString(), value.toString());
})
// 发送消息
.to("topic-out");
return stream;
}
}

复制代码