快捷導(dǎo)航

基于Vue實(shí)現(xiàn)本地消息隊(duì)列MQ的示例詳解

更新時(shí)間：2024年02月27日 08:43:32 作者：大陽(yáng)光男孩

這篇文章為大家詳細(xì)主要介紹了如何基于Vue實(shí)現(xiàn)本地消息隊(duì)列MQ, 讓消息延遲消費(fèi)或者做緩存,文中的示例代碼講解詳細(xì),需要的可以參考一下

MQ功能實(shí)現(xiàn)的具體代碼(TsMQ.ts)：

import { v4 as uuidx } from 'uuid';
 
import emitter from '@/utils/mitt'
 
class Message {
 
    // 過(guò)期時(shí)間，0表示馬上就消費(fèi)
    exp: number;
 
    // 消費(fèi)標(biāo)識(shí)，避免重復(fù)消費(fèi)
    tag : string;
 
    // 消息體
    body : any;
 
    constructor( exp: number ,  body : any ) {
        if (exp == 0) {
            this.exp = 0;
        }else {
            this.exp = Date.now().valueOf() + exp;
        }
        this.body = body;
        this.tag = uuidx();
    }
 
}
 
 
export default class TsMQ {
 
    static tsMQ : TsMQ;
 
    repository : Map<string, any[]>;
 
    /*
      獲取單列的MQ對(duì)象
     */
    static getInstance() : TsMQ {
        if(this.tsMQ == null) {
            this.tsMQ = new TsMQ();
        }
        return this.tsMQ;
    }
 
    constructor() {
        this.repository = new Map<string,any[]>();
        // 把消息放入Mitt進(jìn)行推送
        setInterval(()=> {
            Array.from(this.repository.keys()).forEach( key => {
               let poll = this.repository.get(key) as any[];
               if(poll.length > 0) {
                   poll.forEach( item => {
                       if (item.exp == 0 || item.exp <= Date.now().valueOf() - 100) {
                           emitter.emit(key,item.body);
                           let single : any[] = this.repository.get(key) as any[];
                           single = single.filter(dispose => {
                               return dispose.tag !== item.tag;
                           });
                           this.repository.set(key,single);
                       }
                   });
               }
            });
        },100)
    }
 
 
 
    /*
      * @describe 放消息入隊(duì)列
      * @param queue : string 隊(duì)列名稱
      * @param exp : number 消息消費(fèi)時(shí)間
      * @param  message : any 消息體
     */
    pushMessage( queue : string , exp : number,  message : any ) {
        if(this.repository.has(queue)) {
            let single : any[] = this.repository.get(queue) as any[];
            single.push(new Message(exp,message));
            this.repository.set(queue,single);
        }else {
            let temp = [];
            temp.push(new Message(exp,message));
            this.repository.set(queue,temp);
        }
    }
 
 
    /*
     * 直接獲取消息，可以配合做本地緩存
     * 就要去掉constructor的輪詢推送
     */
    takeMessage( queue : string ) : any {
        let single : any[] = this.repository.get(queue) as any[];
        if( single && single.length > 0) {
            let message = single.shift();
            this.repository.set(queue,single);
            return message;
        }else {
            return '隊(duì)列沒(méi)有消息!';
        }
    }
 
}

提示：其中需要用到三方的uuid和mitt,然后要消息持久化可以用到pinia來(lái)讓消息持久化,本案列沒(méi)有采用持久化

uuid的三方包安裝命令

npm install uuid

使用方式：

<script setup lang="ts">
   import TsMQ from '@/utils/TsMQ'
   import emitter from '@/utils/mitt'
   let tsMQ : TsMQ = TsMQ.getInstance();
 
   //訂閱消息
   emitter.on("HelloWord",e => {
     console.log(`收到消息:${JSON.stringify(e)}\n消息時(shí)間:${new Date().toLocaleString()}`)
   })
 
   //投送消息
   function tsMQs() {
       console.log(`M2投遞時(shí)間：${new Date().toLocaleString()}`)
       tsMQ.pushMessage("HelloWord",1000 * 20,{ name : 'M2', second:20 });
       tsMQ.pushMessage("HelloWord",1000 * 3,{ name : 'M1', second:3 });
   // MQ功能實(shí)現(xiàn)的具體代碼，提示：其中需要用到三方的uuid和mitt,然后要消息持久化可以用到pinia來(lái)讓消息持久化，本案列沒(méi)有采用持久化
   }
 
   function takeMQs() {
       console.log(tsMQ.takeMessage('1'));
   }
 
 
</script>
 
<template>
  <div id="app" style="display: flex;flex-direction: column;justify-content: center;height: 100%;width: 100%;margin: 0;padding: 0">
    <span @click="tsMQs">MQ投送</span>
    <span @click="takeMQs">MQ獲取</span>
  </div>
</template>
 
<style scoped>
 
</style>

效果：