SingleOutputStreamOperator<List<JSONObject>> applyDataStream = dataStreamSource.keyBy(value -> { String[] splitValues = value.getString("key").toString().split(":"); return splitValues[2]; }).window(TumblingProcessingTimeWindows.of(Time.seconds(2))).apply(new WindowFunction<JSONObject, List<JSONObject>, String, TimeWindow>() { @Override public void apply(String key, TimeWindow timeWindow, Iterable<JSONObject> input, Collector<List<JSONObject>> out) throws Exception { List<JSONObject> output = new ArrayList<>(); input.forEach(s -> { // String topicName = ModelUtil.getConfigValue("ods_kafka_all_data_topicName") + "_" + key; //主题名称 String value = s.getString("value"); TranslateJson translateJson = JSON.parseObject(value, TranslateJson.class); if ("read".equals(translateJson.getType()) || "c".equals(translateJson.getType())) {//新增的数据才同步 使用这种写法优化上面代码
时间: 2024-01-12 12:02:26 浏览: 140
java开发-com.alibaba.fastjson.JSONObject
5星 · 资源好评率100%
您可以使用 Lambda 表达式来优化上面的代码,示例代码如下:
```
SingleOutputStreamOperator<List<JSONObject>> applyDataStream = dataStreamSource
.keyBy(value -> value.getString("key").toString().split(":")[2])
.window(TumblingProcessingTimeWindows.of(Time.seconds(2)))
.apply((String key, TimeWindow timeWindow, Iterable<JSONObject> input, Collector<List<JSONObject>> out) -> {
List<JSONObject> output = new ArrayList<>();
input.forEach(s -> {
String value = s.getString("value");
TranslateJson translateJson = JSON.parseObject(value, TranslateJson.class);
if ("read".equals(translateJson.getType()) || "c".equals(translateJson.getType())) {
// 新增的数据才同步
// String topicName = ModelUtil.getConfigValue("ods_kafka_all_data_topicName") + "_" + key; //主题名称
output.add(s);
}
});
out.collect(output);
});
```
使用 Lambda 表达式可以使代码更加简洁易读,同时也可以提高代码的执行效率。
阅读全文