List<JSONObject> not_03_list = jsonObjectList.stream().filter(json -> !json.getString("commodityType").equals("PT03")).collect(Collectors.toList());group

这段代码使用了 `stream()` 方法对 `jsonObjectList` 进行了流操作，通过 `filter()` 方法过滤出 `commodityType` 不等于 `"PT03"` 的元素，最后使用 `collect()` 方法将符合条件的元素收集到一个 `List<JSONObject>` 中。

具体来说，这段代码中的 `filter()` 方法使用了 Lambda 表达式，它的参数是 `json`，表示流中的每个元素，然后通过 `!json.getString("commodityType").equals("PT03")` 的方式过滤出 `commodityType` 属性值不等于 `"PT03"` 的元素。最后使用 `collect()` 方法将符合条件的元素收集到一个 `List<JSONObject>` 中，并赋值给 `not_03_list` 变量。

这个流操作可以简化代码，并提高代码的可读性和效率。使用流操作可以将数据处理过程分解为多个步骤，每个步骤都可以使用不同的方法来实现，使得代码更加灵活和可维护。而且流操作还可以充分利用多核处理器的计算能力，提高处理效率。

相关问题

List<JSONObject> not_03_list = jsonObjectList.stream().filter(json -> !json.getString("commodityType").equals("PT03")).collect(Collectors.toList());根据某个字段分组

要根据某个字段对 `List<JSONObject>` 进行分组，可以使用 `Collectors.groupingBy()` 方法来实现。

例如，假设您有一个 `List<JSONObject>` 对象，并且希望根据 `commodityType` 字段进行分组，可以这样写：

Map<String, List<JSONObject>> groupedMap = jsonObjectList.stream()
        .collect(Collectors.groupingBy(json -> json.getString("commodityType")));

// 输出每个分组中的元素数量
groupedMap.forEach((key, value) -> {
    System.out.println("Commodity type: " + key + ", count: " + value.size());
});

在这个例子中，`groupingBy()` 方法的参数是一个 `Function`，用于将每个元素映射为一个分组的 key。这里我们使用 `json -> json.getString("commodityType")` 的 Lambda 表达式来将每个元素的 `commodityType` 字段作为 key，然后使用 `Map<String, List<JSONObject>>` 类型来存储分组结果。

最后我们使用 `forEach()` 方法遍历每个分组，输出分组的 key 和 value 的大小。

请注意，如果 `commodityType` 字段可能为 `null` 或不存在，建议在使用 `getString()` 方法之前先使用 `containsKey()` 方法或 `getOrDefault()` 方法进行判断，避免出现空指针异常。

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())) {//新增的数据才同步 使用这种写法优化上面代码

您可以使用 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 表达式可以使代码更加简洁易读，同时也可以提高代码的执行效率。