flink重温笔记（十三）： flink 高级特性和新特性（2）——ProcessFunction API 和 双流 join

帮助类具体代码

import { onMounted, ref } from "vue";
import emitter from '@/utils/mitt'

interface Prop {
    exp: number;
    key: string;
    data: any;
    tag?: any;
}



export default function(){

    // 缓存的仓库
    let cacheStore  = ref( {} as Map< string, Prop >);

    // 用于本地排名
    let ranking = ref( {} as Map< string, any[] > );

    // 延迟队列
    let delayMsg = ref( {} as Map< string, any[] > );

    // 用户签到打卡
    let sign = ref( {} as Map< string, any[] > );

    // 限流
    let limiting = ref( {} as Map< string, number> );


    function limit( key : string ) {
        if (limiting.value.has(key)) {
            let temp = limiting.value.get(key) as number;
            limiting.value.set(key,temp + 1);
        }else {
            limiting.value.set(key,1);
        }
    }

    function takeLimit( key : string ) : number {
        return limiting.value.get(key) as number;
    }

    function delLimit( key : string ) {
        limiting.value.delete(key);
    }

    // 放入打卡的数据
    function punchSetting( key : string , signInData : number[]) {
        sign.value.set(key,signInData);
    }


    // 打卡
    function startSign( key : string ,  day : number ): boolean {
        let signInData = sign.value.get(key) as any[];
        if (signInData) {
            signInData.splice(day - 1,1,1);
            return true;
        }
        return false;
    }

    // 返回连续打卡的天数
    function continuousClocking( key : string ) : number {
       if(sign.value.size > 0) {
           let signInData = sign.value.get(key) as any[];
           if (signInData) {
               let day = 1;
               let continuous : any[] = [];
               signInData.forEach( (value, index) => {
                   if(signInData[index] === signInData[index+1] && value ==1) {
                       day++;
                   }else {
                       continuous.push(day);
                       day = 1;
                   }
               });
               return Math.max.apply(null,continuous);
           }
       }
       return 0;
    }

    function putDelayMsg( prop : Prop ) {
        prop.tag = getUUID();
        if (delayMsg.value.has(prop.key)) {
           let delayMsgs = delayMsg.value.get(prop.key) as any[];
           delayMsgs.push(prop);
           delayMsg.value.set(prop.key,delayMsgs);
        }else {
            let temp = [];
            temp.push(prop);
            delayMsg.value.set(prop.key,temp);
        }
    }

    onMounted(() => {
       cacheStore.value = new Map();
       ranking.value = new Map();
       delayMsg.value = new Map();
       sign.value = new Map();
       // 轮询清除过期的缓存
       setInterval(() => {
            if(cacheStore.value.size > 0) {
                Array.from(cacheStore.value.keys()).forEach( key => {
                    let data = cacheStore.value.get(key);
                    if (data) {
                        let nowDate = new Date().valueOf();
                        if (data.exp < nowDate && data.exp != -1) {
                            cacheStore.value.delete(key);
                        }
                    }
                });
            }
       },100);
        // 发送过期消息让Mitt进行通知 emitter.on('xxx', e => {  console.log(e) } )
        setInterval(() => {
            if(delayMsg.value.size > 0) {
                Array.from(delayMsg.value.keys()).forEach( key => {
                    let temp = delayMsg.value.get(key) as any[];
                    temp.forEach( item => {
                        let nowDate = new Date().valueOf();
                        if( item.exp < nowDate - 100 ) {
                            emitter.emit(item.key, item.data);
                            temp = temp.filter( dispose  => dispose .tag != item.tag);
                            delayMsg.value.set(item.key,temp);
                        }
                    });
                })
            }
        },100);
    });


    function putRanking( key : string , data : any) {
        if(ranking.value.has(key)) {
           let rank  = ranking.value.get(key) as any[];
           let temp = rank.find(item => item.data === data);
           if(temp) {
               let newVal =  rank.filter(item => item.data !== data );
               newVal.push({ data: data.toString() , score : temp.score + 1 });
               ranking.value.set(key,newVal);
           }else {
               rank.push({ data: data.toString() , score : 1 });
               ranking.value.set(key,rank);
           }
        }else {
           let temp = [] as any[];
           temp.push({ data: data.toString() , score: 1 });
           ranking.value.set(key,temp);
        }
    }

    function takeRanking( key : string , total : number , desc : boolean): any[] {
        let rank = ranking.value.get(key) as any[];
        if(desc) {
            rank.sort( ( a , b ) => b.score - a.score );
        }else {
            rank.sort( ( a , b ) => a.score - b.score );
        }
        return rank.slice(0, total != 1 ? total - 1 : total);
    }

    // 放入缓存
    function putCache( prop : Prop ) {
        cacheStore.value.set(prop.key,prop);
    }


    // 获取缓存
    function takeCache<T>( key : string ) : any {
        return cacheStore.value.get(key) as T;
    }

    // 加锁
    function lock( prop : Prop ) : boolean {
        if (getLock(prop.key)) {
            return false;
        }
        cacheStore.value.set(prop.key,prop);
        return true;
    }


    // 获取锁
    function getLock( key : string ) : boolean {
        return cacheStore.value.has(key);
    }

    // 解锁
    function unLock( key : string , id : any) {
        let lock = cacheStore.value.get(key);
        if( lock ) {
            if( lock.data == id ) {
                cacheStore.value.delete(key);
            }
        }
    }


    // 全局id
    function getUUID () {
        return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function (c) {
            const r = Math.random() * 16 | 0
            const v = c === 'x' ? r : (r & 0x3 | 0x8)
            return v.toString(16)
        });
    }

    return { takeCache , putCache , unLock , lock , getLock , putRanking , takeRanking , getUUID , putDelayMsg , startSign , punchSetting , continuousClocking }
}

使用代码

<template>
  <div style="display: flex;justify-content: center;align-items: center;height: 120px;gap: 12px">
    <button @click="getCache">获取缓存</button>
    <button @click="getSearch">获取前三搜索</button>
    <button @click="makeOrder">生成订单</button>
    <button @click="initSign">初始打卡数据</button>
    <button @click="simulationSign">模拟打卡</button>
    <button>获取练习打卡{{continuousClocking('3')}}</button>
  </div>
</template>

<style scoped>
    @import 'animate.css';
    .ske {
      display: flex;
      justify-content: center;
      align-items: center;
      animation: hinge;
      animation-duration: 3s;
    }
</style>

<script setup lang="ts">


    import { onMounted } from 'vue';

    import emitter from '@/utils/mitt'

    import CacheHelp from '@/hooks/CacheHelp'

    const { putCache , takeCache , putRanking , takeRanking ,  putDelayMsg , punchSetting , continuousClocking , startSign } = CacheHelp();

    onMounted(() => {
      putCache({ key: 'UserInfo' , data : { name: 'Jack' , age : 12 } , exp: new Date().valueOf() + 3000 });
      putRanking("word","成都");
      putRanking("word","北京");
      putRanking("word","成都");
      putRanking("word","重庆");
      putRanking("word","武汉");
      putRanking("word","北京");
      putRanking("word","成都");
      putRanking("word","上海");
    });

    function simulationSign() {
      startSign('3',2);
      startSign('3',3);
      startSign('3',4);
      startSign('3',12);
      startSign('3',13);
      startSign('3',14);
      startSign('3',15);
      startSign('3',16);
      startSign('3',18);
      startSign('3',31);
    }



    function getCache() {
      console.log(`获取的缓存：${JSON.stringify(takeCache("UserInfo"))}`);
    }

    function getMonthDays( year : number, month : number ) : number {
      let days = [31,28,31,30,31,30,31,31,30,31,30,31]
      if ( ( year % 4 === 0 ) && ( year % 100 !== 0 || year % 400 === 0) ) {
        days[1] = 29
      }
      return days[month - 1];
    }

    function initSign() {
      let signData = []
      for (let i = 0; i < getMonthDays(2024,3); i++) {
        signData.push(0);
      }
      punchSetting('3',signData);
    }

    function getSearch() {
      console.log(`排名考前的缓存：${JSON.stringify(takeRanking('word',3,true))}`);
    }

    function makeOrder() {
        console.log(`发送的时间：${new Date().toLocaleTimeString()}`);
        putDelayMsg({ key: 'orderDelay' , data: { price :12.9 , id : 1 , name : '大衣' } , exp : new Date().valueOf() + 10000 });
        putDelayMsg({ key: 'orderDelay' , data: { price :15.9 , id : 2 , name : '短袖' } , exp : new Date().valueOf() + 10000 });
    }

    emitter.on("orderDelay",e => {
      console.log(`收到延迟消息：${JSON.stringify(e)}\t接收时间：${new Date().toLocaleTimeString()}`)
    })



</script>

效果