栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Java

flink消费kafka消息,自定义keyselector

Java 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

flink消费kafka消息,自定义keyselector

为了能达到在不同的keyby/window中,能按照自定义的业务要求,把相同的消息分到同一个消费窗口,需要自定义keyselector,如本例中,我想要按照消息体中的url字段,把相同url字段的消息,用同一个线程来消费:写法为:

//通过自定义keyby selector,能保证相同url的新媒体消息,出现在同一个keyby/window窗口
        env.fromSource(newSource, WatermarkStrategy.noWatermarks(), "newPageSource")
                .keyBy((KeySelector) consumerRecord -> {
                    String jsonStr = new String((byte[]) consumerRecord.value(),
                            StandardCharsets.UTF_8);
                    if (StrUtil.isNotBlank(jsonStr) && !"null".equals(jsonStr)) {
                        JSONObject jsonObject = JSONUtil.parseObj(jsonStr);
                        String url = jsonObject.getStr("url");
                        return url;
                    }

                    return "";
                }).addSink(newPageSink).name("newPageSink");

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/1035552.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号