Flink操练（三十一）之状态判断连续上升数秒

　　1、代码逻辑实现packageday03；importorg。apache。flink。api。common。state。ValueState；importorg。apache。flink。api。common。state。ValueStateDescriptor；importorg。apache。flink。api。common。typeinfo。Types；importorg。apache。flink。configuration。Configuration；importorg。apache。flink。streaming。api。environment。StreamExecutionEnvironment；importorg。apache。flink。streaming。api。functions。KeyedProcessFunction；importorg。apache。flink。streaming。api。functions。source。SourceFunction；importorg。apache。flink。util。Collector；importjava。util。Random；整数连续1s上升publicclassOneSecondByContinuousUp｛publicstaticvoidmain（String〔〕args）throwsException｛StreamExecutionEnvironmentenvStreamExecutionEnvironment。getExecutionEnvironment（）；env。setParallelism（1）；env。addSource（newSourceFunctionInteger（）｛privatebooleanrunningtrue；privateRandomrandomnewRandom（）；Overridepublicvoidrun（SourceContextIntegerctx）throwsException｛while（running）｛ctx。collect（random。nextInt（））；Thread。sleep（300L）；｝｝Overridepublicvoidcancel（）｛runningfalse；｝｝）。keyBy（r1）。process（newIntIncreaseAlert（））。print（）；env。execute（）；｝publicstaticclassIntIncreaseAlertextendsKeyedProcessFunctionInteger，Integer，String｛privateValueStateIntegerlastInt；privateValueStateLongtimerTs；Overridepublicvoidopen（Configurationparameters）throwsException｛super。open（parameters）；lastIntgetRuntimeContext（）。getState（newValueStateDescriptorInteger（lastinteger，Types。INT））；timerTsgetRuntimeContext（）。getState（newValueStateDescriptorLong（timer，Types。LONG））；｝OverridepublicvoidprocessElement（Integervalue，Contextctx，CollectorStringout）throwsException｛IntegerprevIntnull；if（lastInt。value（）！null）｛prevIntlastInt。value（）；｝lastInt。update（value）；Longtsnull；if（timerTs。value（）！null）｛tstimerTs。value（）；｝if（prevIntnullvalueprevInt）｛if（ts！null）｛ctx。timerService（）。deleteProcessingTimeTimer（ts）；timerTs。clear（）；｝｝elseif（valueprevInttsnull）｛longoneSecLaterctx。timerService（）。currentProcessingTime（）1000L；ctx。timerService（）。registerProcessingTimeTimer（oneSecLater）；timerTs。update（oneSecLater）；｝｝OverridepublicvoidonTimer（longtimestamp，OnTimerContextctx，CollectorStringout）throwsException｛super。onTimer（timestamp，ctx，out）；out。collect（整数连续1s上升了！）；timerTs。clear（）；｝｝｝2、结果之展示C：ProgramFilesJavajdk1。8。0191binjava。exejavaagent：F：appIntelliJIDEA2019。3。3libideart。jar57967：F：appIntelliJIDEA2019。3。3binDfile。encodingUTF8classpathC：ProgramFilesJavajdk1。8。0191jrelibcharsets。jar；C：ProgramFilesJavajdk1。8。0191jrelibdeploy。jar；C：ProgramFilesJavajdk1。8。0191jrelibextaccessbridge64。jar；C：ProgramFilesJavajdk1。8。0191jrelibextcldrdata。jar；C：ProgramFilesJavajdk1。8。0191jrelibextdnsns。jar；C：ProgramFilesJavajdk1。8。0191jrelibextjaccess。jar；C：ProgramFilesJavajdk1。8。0191jrelibextjfxrt。jar；C：ProgramFilesJavajdk1。8。0191jrelibextlocaledata。jar；C：ProgramFilesJavajdk1。8。0191jrelibextashorn。jar；C：ProgramFilesJavajdk1。8。0191jrelibextsunec。jar；C：ProgramFilesJavajdk1。8。0191jrelibextsunjceprovider。jar；C：ProgramFilesJavajdk1。8。0191jrelibextsunmscapi。jar；C：ProgramFilesJavajdk1。8。0191jrelibextsunpkcs11。jar；C：ProgramFilesJavajdk1。8。0191jrelibextzipfs。jar；C：ProgramFilesJavajdk1。8。0191jrelibjavaws。jar；C：ProgramFilesJavajdk1。8。0191jrelibjce。jar；C：ProgramFilesJavajdk1。8。0191jrelibjfr。jar；C：ProgramFilesJavajdk1。8。0191jrelibjfxswt。jar；C：ProgramFilesJavajdk1。8。0191jrelibjsse。jar；C：ProgramFilesJavajdk1。8。0191jrelibmanagementagent。jar；C：ProgramFilesJavajdk1。8。0191jrelibplugin。jar；C：ProgramFilesJavajdk1。8。0191jrelibresources。jar；C：ProgramFilesJavajdk1。8。0191jrelibrt。jar；D：bigDatabigDatalearnFlinklearnargetclasses；C：UsersAdministrator。m2repositoryorgapacheflinkflinkjava1。13。0flinkjava1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkcore1。13。0flinkcore1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkannotations1。13。0flinkannotations1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkmetricscore1。13。0flinkmetricscore1。13。0。jar；C：UsersAdministrator。m2repositorycomesotericsoftwarekryokryo2。24。0kryo2。24。0。jar；C：UsersAdministrator。m2repositorycomesotericsoftwareminlogminlog1。2minlog1。2。jar；C：UsersAdministrator。m2repositoryorgobjenesisobjenesis2。1objenesis2。1。jar；C：UsersAdministrator。m2repositorycommonscollectionscommonscollections3。2。2commonscollections3。2。2。jar；C：UsersAdministrator。m2repositoryorgapachecommonscommonscompress1。20commonscompress1。20。jar；C：UsersAdministrator。m2repositoryorgapachecommonscommonslang33。3。2commonslang33。3。2。jar；C：UsersAdministrator。m2repositoryorgapachecommonscommonsmath33。5commonsmath33。5。jar；C：UsersAdministrator。m2repositoryorgslf4jslf4japi1。7。15slf4japi1。7。15。jar；C：UsersAdministrator。m2repositorycomgooglecodefindbugsjsr3051。3。9jsr3051。3。9。jar；C：UsersAdministrator。m2repositoryorgapacheflinkforceshading1。13。0forceshading1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkstreamingjava2。121。13。0flinkstreamingjava2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkfilesinkcommon1。13。0flinkfilesinkcommon1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkruntime2。121。13。0flinkruntime2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkqueryablestateclientjava1。13。0flinkqueryablestateclientjava1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkhadoopfs1。13。0flinkhadoopfs1。13。0。jar；C：UsersAdministrator。m2repositorycommonsiocommonsio2。7commonsio2。7。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkshadednetty4。1。49。Final13。0flinkshadednetty4。1。49。Final13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkshadedjackson2。12。113。0flinkshadedjackson2。12。113。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkshadedzookeeper33。4。1413。0flinkshadedzookeeper33。4。1413。0。jar；C：UsersAdministrator。m2repositoryorgjavassistjavassist3。24。0GAjavassist3。24。0GA。jar；C：UsersAdministrator。m2repositorycomypesafeakkaakkaactor2。122。5。21akkaactor2。122。5。21。jar；C：UsersAdministrator。m2repositorycomypesafeconfig1。3。3config1。3。3。jar；C：UsersAdministrator。m2repositoryorgscalalangmodulesscalajava8compat2。12。8。0scalajava8compat2。120。8。0。jar；C：UsersAdministrator。m2repositorycomypesafeakkaakkastream2。122。5。21akkastream2。122。5。21。jar；C：UsersAdministrator。m2repositoryorgreactivestreamsreactivestreams1。0。2reactivestreams1。0。2。jar；C：UsersAdministrator。m2repositorycomypesafesslconfigcore2。12。3。7sslconfigcore2。120。3。7。jar；C：UsersAdministrator。m2repositoryorgscalalangmodulesscalaparsercombinators2。121。1。1scalaparsercombinators2。121。1。1。jar；C：UsersAdministrator。m2repositorycomypesafeakkaakkaprotobuf2。122。5。21akkaprotobuf2。122。5。21。jar；C：UsersAdministrator。m2repositorycomypesafeakkaakkaslf4j2。122。5。21akkaslf4j2。122。5。21。jar；C：UsersAdministrator。m2repositoryorgclappergrizzledslf4j2。121。3。2grizzledslf4j2。121。3。2。jar；C：UsersAdministrator。m2repositorycomgithubscoptscopt2。123。5。0scopt2。123。5。0。jar；C：UsersAdministrator。m2repositoryorgxerialsnappysnappyjava1。1。8。3snappyjava1。1。8。3。jar；C：UsersAdministrator。m2repositorycomwitterchill2。12。7。6chill2。120。7。6。jar；C：UsersAdministrator。m2repositorycomwitterchilljava。7。6chilljava0。7。6。jar；C：UsersAdministrator。m2repositoryorglz4lz4java1。6。0lz4java1。6。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkshadedguava18。013。0flinkshadedguava18。013。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkclients2。121。13。0flinkclients2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkoptimizer2。121。13。0flinkoptimizer2。121。13。0。jar；C：UsersAdministrator。m2repositorycommonsclicommonscli1。3。1commonscli1。3。1。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinktableapijavabridge2。121。13。0flinktableapijavabridge2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinktableapijava1。13。0flinktableapijava1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinktableplannerblink2。121。13。0flinktableplannerblink2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinktableapiscala2。121。13。0flinktableapiscala2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinktableapiscalabridge2。121。13。0flinktableapiscalabridge2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinktableruntimeblink2。121。13。0flinktableruntimeblink2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgcodehausjaninojanino3。0。11janino3。0。11。jar；C：UsersAdministrator。m2repositoryorgcodehausjaninocommonscompiler3。0。11commonscompiler3。0。11。jar；C：UsersAdministrator。m2repositoryorgapachecalciteavaticaavaticacore1。17。0avaticacore1。17。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkstreamingscala2。121。13。0flinkstreamingscala2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkscala2。121。13。0flinkscala2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgscalalangscalareflect2。12。7scalareflect2。12。7。jar；C：UsersAdministrator。m2repositoryorgscalalangscalalibrary2。12。7scalalibrary2。12。7。jar；C：UsersAdministrator。m2repositoryorgscalalangscalacompiler2。12。7scalacompiler2。12。7。jar；C：UsersAdministrator。m2repositoryorgscalalangmodulesscalaxml2。121。0。6scalaxml2。121。0。6。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinktablecommon1。13。0flinktablecommon1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkconnectorfiles1。13。0flinkconnectorfiles1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkshadedasm77。113。0flinkshadedasm77。113。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkcep2。121。13。0flinkcep2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkcsv1。13。0flinkcsv1。13。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkconnectorkafka2。121。13。0flinkconnectorkafka2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgapachekafkakafkaclients2。4。1kafkaclients2。4。1。jar；C：UsersAdministrator。m2repositorycomgithublubenzstdjni1。4。31zstdjni1。4。31。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkconnectorbase1。13。0flinkconnectorbase1。13。0。jar；C：UsersAdministrator。m2repositoryorgapachebahirflinkconnectorredis2。111。0flinkconnectorredis2。111。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkstreamingjava2。111。2。0flinkstreamingjava2。111。2。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkruntime2。111。2。0flinkruntime2。111。2。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkshadedhadoop21。2。0flinkshadedhadoop21。2。0。jar；C：UsersAdministrator。m2repositoryorgukaanixz1。0xz1。0。jar；C：UsersAdministrator。m2repositoryxmlencxmlenc。52xmlenc0。52。jar；C：UsersAdministrator。m2repositorycommonscodeccommonscodec1。4commonscodec1。4。jar；C：UsersAdministrator。m2repositorycommonsnetcommonsnet3。1commonsnet3。1。jar；C：UsersAdministrator。m2repositoryjavaxservletservletapi2。5servletapi2。5。jar；C：UsersAdministrator。m2repositoryorgmortbayjettyjettyutil6。1。26jettyutil6。1。26。jar；C：UsersAdministrator。m2repositorycomsunjerseyjerseycore1。9jerseycore1。9。jar；C：UsersAdministrator。m2repositorycommonselcommonsel1。0commonsel1。0。jar；C：UsersAdministrator。m2repositorycommonsloggingcommonslogging1。1。3commonslogging1。1。3。jar；C：UsersAdministrator。m2repositorycomjamesmurtyutilsjavaxmlbuilder。4javaxmlbuilder0。4。jar；C：UsersAdministrator。m2repositorycommonslangcommonslang2。6commonslang2。6。jar；C：UsersAdministrator。m2repositorycommonsconfigurationcommonsconfiguration1。7commonsconfiguration1。7。jar；C：UsersAdministrator。m2repositorycommonsdigestercommonsdigester1。8。1commonsdigester1。8。1。jar；C：UsersAdministrator。m2repositoryorgcodehausjacksonjacksoncoreasl1。8。8jacksoncoreasl1。8。8。jar；C：UsersAdministrator。m2repositoryorgcodehausjacksonjacksonmapperasl1。8。8jacksonmapperasl1。8。8。jar；C：UsersAdministrator。m2repositoryorgapacheavroavro1。7。7avro1。7。7。jar；C：UsersAdministrator。m2repositorycomhoughtworksparanamerparanamer2。3paranamer2。3。jar；C：UsersAdministrator。m2repositorycomjcraftjsch。1。42jsch0。1。42。jar；C：UsersAdministrator。m2repositorycommonsbeanutilscommonsbeanutilsbeancollections1。8。3commonsbeanutilsbeancollections1。8。3。jar；C：UsersAdministrator。m2repositorycommonsdaemoncommonsdaemon1。0。13commonsdaemon1。0。13。jar；C：UsersAdministrator。m2repositoryjavaxxmlbindjaxbapi2。2。2jaxbapi2。2。2。jar；C：UsersAdministrator。m2repositoryjavaxxmlstreamstaxapi1。02staxapi1。02。jar；C：UsersAdministrator。m2repositoryjavaxactivationactivation1。1activation1。1。jar；C：UsersAdministrator。m2repositoryioettyettyall4。0。27。Finalettyall4。0。27。Final。jar；C：UsersAdministrator。m2repositorycomdataartisansflakkaactor2。112。3customflakkaactor2。112。3custom。jar；C：UsersAdministrator。m2repositorycomdataartisansflakkaremote2。112。3customflakkaremote2。112。3custom。jar；C：UsersAdministrator。m2repositoryioettyetty3。8。0。Finaletty3。8。0。Final。jar；C：UsersAdministrator。m2repositoryorguncommonsmathsuncommonsmaths1。2。2auncommonsmaths1。2。2a。jar；C：UsersAdministrator。m2repositorycomdataartisansflakkaslf4j2。112。3customflakkaslf4j2。112。3custom。jar；C：UsersAdministrator。m2repositoryorgclappergrizzledslf4j2。111。0。2grizzledslf4j2。111。0。2。jar；C：UsersAdministrator。m2repositorycomgithubscoptscopt2。113。2。0scopt2。113。2。0。jar；C：UsersAdministrator。m2repositorycomfasterxmljacksoncorejacksoncore2。7。4jacksoncore2。7。4。jar；C：UsersAdministrator。m2repositorycomfasterxmljacksoncorejacksondatabind2。7。4jacksondatabind2。7。4。jar；C：UsersAdministrator。m2repositorycomfasterxmljacksoncorejacksonannotations2。7。0jacksonannotations2。7。0。jar；C：UsersAdministrator。m2repositoryorgapachezookeeperzookeeper3。4。6zookeeper3。4。6。jar；C：UsersAdministrator。m2repositoryjlinejline。9。94jline0。9。94。jar；C：UsersAdministrator。m2repositoryjunitjunit3。8。1junit3。8。1。jar；C：UsersAdministrator。m2repositorycomwitterchill2。11。7。4chill2。110。7。4。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkclients2。111。2。0flinkclients2。111。2。0。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkoptimizer2。111。2。0flinkoptimizer2。111。2。0。jar；C：UsersAdministrator。m2repositoryorgapacheslingorg。apache。sling。commons。json2。0。6org。apache。sling。commons。json2。0。6。jar；C：UsersAdministrator。m2repositorymysqlmysqlconnectorjava8。0。21mysqlconnectorjava8。0。21。jar；C：UsersAdministrator。m2repositorycomgoogleprotobufprotobufjava3。11。4protobufjava3。11。4。jar；C：UsersAdministrator。m2repositoryorgapacheflinkflinkconnectorjdbc2。121。13。0flinkconnectorjdbc2。121。13。0。jar；C：UsersAdministrator。m2repositoryorgslf4jslf4jlog4j121。7。30slf4jlog4j121。7。30。jar；C：UsersAdministrator。m2repositorylog4jlog4j1。2。17log4j1。2。17。jar；C：UsersAdministrator。m2repositoryorgapachelogginglog4jlog4jtoslf4j2。14。0log4jtoslf4j2。14。0。jar；C：UsersAdministrator。m2repositoryorgapachelogginglog4jlog4japi2。14。0log4japi2。14。0。jar；C：UsersAdministrator。m2repositoryredisclientsjedis2。9。0jedis2。9。0。jar；C：UsersAdministrator。m2repositoryorgapachecommonscommonspool22。4。2commonspool22。4。2。jar；C：UsersAdministrator。m2repositorycomgooglecodegsongson2。8。5gson2。8。5。jarday03。OneSecondByContinuousUplog4j：WARNNoappenderscouldbefoundforlogger（org。apache。flink。api。java。ClosureCleaner）。log4j：WARNPleaseinitializethelog4jsystemproperly。log4j：WARNSeehttp：logging。apache。orglog4j1。2faq。htmlnoconfigformoreinfo。整数连续1s上升了！