WebSocket中心跳檢測(cè)與斷開重連機(jī)制詳解

更新時(shí)間：2024年01月10日 08:18:34 作者：hktk_wb

心跳檢測(cè)是指在WebSocket連接過程中定時(shí)向服務(wù)端發(fā)送和接收心跳消息,來確定當(dāng)前連接是否是正常狀態(tài)的檢測(cè)機(jī)制,斷開重連是指在WebSocket不正常斷開連接后,進(jìn)行重新連接的策略,下面我們就來看看這二者的具體實(shí)現(xiàn)吧

心跳檢測(cè)

在長(zhǎng)時(shí)間WebSocket連接過程中，容易因?yàn)榫W(wǎng)絡(luò)或其他原因?qū)е?code>WebSocket斷開而沒有觸發(fā)ws.onclose事件，導(dǎo)致無法獲取WebSocket連接失敗的狀態(tài)。

利用心跳檢測(cè)可以預(yù)防這一問題。心跳檢測(cè)是指在WebSocket連接過程中定時(shí)向服務(wù)端發(fā)送和接收心跳消息，來確定當(dāng)前連接是否是正常狀態(tài)的檢測(cè)機(jī)制。

斷開重連

斷開重連是指在WebSocket不正常斷開連接后，進(jìn)行重新連接的策略。

在WebSocket連接斷開后，立馬進(jìn)行重新連接。如果連接失敗，則間隔一定時(shí)間后再嘗試重連，直到連接成功或者達(dá)到最大重連次數(shù)。

消息訂閱

在WebSocket使用過程中，可能會(huì)在不同的頁面或邏輯中需要訂閱全部或部分WebSocket消息，利用消息訂閱可以讓使用者在不同頁面中訂閱獲取相關(guān)的消息信息。

代碼實(shí)現(xiàn)

定義WebSocket內(nèi)部參數(shù)

this.url = url //WebSocket連接地址
this.ws = null //WebSocket連接對(duì)象
this.heartEnabled = heartEnabled //是否開啟心跳
this.heartInterval = 10000 //心跳間隔時(shí)間
this.heartTimeout = 5000 //心跳超時(shí)時(shí)間
this.lockReconnect = false //是否禁止重連
this.heartTimer = null //心跳定時(shí)器
this.serverTimer = null //服務(wù)器超時(shí)定時(shí)器
this.reconnectCount = 0 //重連次數(shù)
this.maxReconnectCount = 5 //最大重連次數(shù)
this.observers = [] //消息訂閱者列表
this.waitingMessages = [] //待執(zhí)行命令列表

實(shí)現(xiàn)WebSocket連接

  //WebSocket連接
  connect() {
    this.ws = new WebSocket(this.url)

    this.ws.onopen = () => {
      this.reconnectCount = 0 // 重置重連次數(shù)
      this.heartEnabled && this.start() // 開啟心跳

      // 發(fā)送所有等待發(fā)送的信息
      const length = this.waitingMessages.length
      for (let i = 0; i < length; ++i) {
        const message = this.waitingMessages.shift()
        this.send(message)
      }
    }

    this.ws.onclose = (event) => {
      console.log('WebSocket closed:', event)
      this.reconnect() //這里可根據(jù)具體情況判定是否重新連接
    }

    this.ws.onerror = (error) => {
      console.log('WebSocket error:', error)
      this.reconnect()
    }

    this.ws.onmessage = (event) => {
      //收到心跳信息則重置心跳，收到其他信息則觸發(fā)回調(diào)
      if (event.data === 'pong') {
        this.start()
      } else {
        this.observers.forEach((observer) => {
          //如需要，這里可根據(jù)observer的訂閱type來判斷是否回調(diào)
          observer.callback(event.data)
        })
      }
    }
  }

消息訂閱及發(fā)送

  //發(fā)送信息
  send(message) {
    //發(fā)送信息時(shí)若WebSocket還未連接，則將信息放入待發(fā)送信息中等待連接成功后發(fā)送
    if (this.onReady() !== WebSocket.OPEN) {
      this.waitingMessages.push(message)
      return this
    }

    this.ws.send(message)
    return this
  }

  //訂閱WebSocket信息
  onObserve(callback, type = 'all') {
    const observer = { type, callback }
    this.observers.push(observer)

    return observer
  }

  //取消訂閱信息
  cancelObserve(cancelObserver) {
    this.observers.forEach((observer, index) => {
      if (cancelObserver === observer) {
        this.observers.splice(index, 1)
      }
    })
  }

  //WebSocket連接狀態(tài)
  onReady() {
    return this.ws.readyState
  }

心跳檢測(cè)

  //開啟心跳
  start() {
    this.reset()

    this.heartTimer = setTimeout(() => {
      this.send('ping')

      //達(dá)到心跳超時(shí)時(shí)間還沒有返回心跳信息，則認(rèn)為連接斷開，關(guān)閉WebSocket并重連
      this.serverTimer = setTimeout(() => {
        this.ws.close()
      }, this.heartTimeout)
    }, this.heartInterval)
  }

  //重置心跳定時(shí)器/服務(wù)超時(shí)定時(shí)器
  reset() {
    this.heartTimer && clearTimeout(this.heartTimer)

    this.serverTimer && clearTimeout(this.serverTimer)
  }

斷開重連

  //重連
  reconnect() {
    // 設(shè)置lockReconnect變量避免重復(fù)連接
    if (this.lockReconnect || this.reconnectCount >= this.maxReconnectCount) return
    this.lockReconnect = true

    this.reconnectCount++ //重連次數(shù)+1

    setTimeout(() => {
      this.connect()
      this.lockReconnect = false
    }, 1000 * this.reconnectCount) //重連次數(shù)越多，延時(shí)越久
  }

完整代碼 js

class SocketConnect {
  constructor(url = 'ws://127.0.0.1:8080', heartEnabled = true) {
    this.url = url //WebSocket連接地址
    this.ws = null //WebSocket連接對(duì)象
    this.heartEnabled = heartEnabled //是否開啟心跳
    this.heartInterval = 10000 //心跳間隔時(shí)間
    this.heartTimeout = 5000 //心跳超時(shí)時(shí)間
    this.lockReconnect = false //是否禁止重連
    this.heartTimer = null //心跳定時(shí)器
    this.serverTimer = null //服務(wù)器超時(shí)定時(shí)器
    this.reconnectCount = 0 //重連次數(shù)
    this.maxReconnectCount = 5 //最大重連次數(shù)
    this.observers = [] //消息訂閱者列表
    this.waitingMessages = [] //待執(zhí)行命令列表

    this.connect()
  }

  //WebSocket連接
  connect() {
    this.ws = new WebSocket(this.url)

    this.ws.onopen = () => {
      this.reconnectCount = 0 // 重置重連次數(shù)
      this.heartEnabled && this.start() // 開啟心跳

      // 發(fā)送所有等待發(fā)送的信息
      const length = this.waitingMessages.length
      for (let i = 0; i < length; ++i) {
        const message = this.waitingMessages.shift()
        this.send(message)
      }
    }

    this.ws.onclose = (event) => {
      console.log('WebSocket closed:', event)
      this.reconnect()
    }

    this.ws.onerror = (error) => {
      console.log('WebSocket error:', error)
      this.reconnect()
    }

    this.ws.onmessage = (event) => {
      //收到心跳信息則重置心跳，收到其他信息則觸發(fā)回調(diào)
      if (event.data === 'pong') {
        this.start()
      } else {
        this.observers.forEach((observer) => {
          observer.callback(event.data)
        })
      }
    }
  }

  //發(fā)送信息
  send(message) {
    //發(fā)送信息時(shí)若WebSocket還未連接，則將信息放入待發(fā)送信息中等待連接成功后發(fā)送
    if (this.onReady() !== WebSocket.OPEN) {
      this.waitingMessages.push(message)
      return this
    }

    this.ws.send(message)
    return this
  }

  //訂閱webSocket信息
  onObserve(callback, type = 'all') {
    const observer = { type, callback }
    this.observers.push(observer)

    return observer
  }

  //取消訂閱信息
  cancelObserve(cancelObserver) {
    this.observers.forEach((observer, index) => {
      if (cancelObserver === observer) {
        this.observers.splice(index, 1)
      }
    })
  }

  //開啟心跳
  start() {
    this.reset()

    this.heartTimer = setTimeout(() => {
      this.send('ping')

      //5秒鐘還沒有返回心跳信息，則認(rèn)為連接斷開，關(guān)閉WebSocket并重連
      this.serverTimer = setTimeout(() => {
        this.ws.close()
      }, this.heartTimeout)
    }, this.heartInterval)
  }

  //重置心跳定時(shí)器/服務(wù)超時(shí)定時(shí)器
  reset() {
    this.heartTimer && clearTimeout(this.heartTimer)

    this.serverTimer && clearTimeout(this.serverTimer)
  }

  //重連
  reconnect() {
    // 設(shè)置lockReconnect變量避免重復(fù)連接
    if (this.lockReconnect || this.reconnectCount >= this.maxReconnectCount) return
    this.lockReconnect = true

    this.reconnectCount++ //重連次數(shù)+1

    setTimeout(() => {
      this.connect()
      this.lockReconnect = false
    }, 1000 * this.reconnectCount) //重連次數(shù)越多，延時(shí)越久
  }

  //WebSocket連接狀態(tài)
  onReady() {
    return this.ws.readyState
  }
}

export default SocketConnect

使用示例

// WebSocket連接
const url = 'ws://127.0.0.1:8080'
const ws = new SocketConnect(url)

// 消息訂閱
const observer = ws.onObserve((data) => {
  console.log('data:', data)
})

// 取消訂閱
ws.cancelObserve(observer)

// 發(fā)送消息
ws.send('hello world')

完整代碼 ts

type ObserverType = {
  type: string
  callback: (event: MessageEvent) => void
}

type MessageDataType = object

class SocketConnect {
  private url: string
  private ws: WebSocket | undefined //WebSocket實(shí)例
  private heartEnabled: boolean //是否開啟心跳
  private heartInterval = 10000 //心跳間隔時(shí)間
  private heartTimeout = 5000 //心跳超時(shí)時(shí)間
  private lockReconnect = false //是否禁止重連
  private heartTimer: NodeJS.Timeout | undefined //心跳定時(shí)器
  private serverTimer: NodeJS.Timeout | undefined //服務(wù)器超時(shí)定時(shí)器
  private reconnectCount = 0 //重連次數(shù)
  private maxReconnectCount = 5 //最大重連次數(shù)
  private observers: ObserverType[] = [] //消息訂閱者列表
  private waitingMessages: string[] = [] //待執(zhí)行命令列表

  constructor(url = 'ws://127.0.0.1:8080', heartEnabled = false) {
    this.url = url
    this.heartEnabled = heartEnabled

    this.connect()
  }

  //WebSocket連接
  connect() {
    this.ws = new WebSocket(this.url)

    this.ws.onopen = () => {
      this.reconnectCount = 0 // 重置重連次數(shù)
      this.heartEnabled && this.start() // 開啟心跳

      // 發(fā)送所有等待發(fā)送的信息
      const length = this.waitingMessages.length
      for (let i = 0; i < length; ++i) {
        const message = this.waitingMessages.shift()
        this.send(message)
      }
    }

    this.ws.onclose = (event) => {
      console.log('WebSocket closed:', event)
      this.reconnect()
    }

    this.ws.onerror = (error) => {
      console.log('WebSocket error:', error)
      this.reconnect()
    }

    this.ws.onmessage = (event: MessageEvent) => {
      //收到心跳信息則重置心跳，收到其他信息則觸發(fā)回調(diào)
      if (event.data === 'pong') {
        this.start()
      } else {
        this.observers.forEach((observer) => {
          observer.callback(event.data)
        })
      }
    }
  }

  //發(fā)送信息
  send(message?: string) {
    if (message) {
      //發(fā)送信息時(shí)若WebSocket還未連接，則將信息放入待發(fā)送信息中等待連接成功后發(fā)送
      if (this.onReady() !== WebSocket.OPEN) {
        this.waitingMessages.push(message)
        return this
      }

      this.ws && this.ws.send(message)
    }

    return this
  }

  //訂閱WebSocket信息
  onObserve(callback: (data: MessageDataType) => void, type = 'all') {
    const observer = { type, callback }
    this.observers.push(observer)

    return observer
  }

  //取消訂閱信息
  cancelObserve(cancelObserver: ObserverType) {
    this.observers.forEach((observer, index) => {
      if (cancelObserver === observer) {
        this.observers.splice(index, 1)
      }
    })
  }

  //開啟心跳
  private start() {
    this.reset()

    this.heartTimer = setTimeout(() => {
      this.send('ping')

      //5秒鐘還沒有返回心跳信息，則認(rèn)為連接斷開，關(guān)閉WebSocket并重連
      this.serverTimer = setTimeout(() => {
        this.ws && this.ws.close()
      }, this.heartTimeout)
    }, this.heartInterval)
  }

  //重連
  private reconnect() {
    // 設(shè)置lockReconnect變量避免重復(fù)連接
    if (this.lockReconnect || this.reconnectCount >= this.maxReconnectCount) return
    this.lockReconnect = true

    this.reconnectCount++ //重連次數(shù)+1

    setTimeout(() => {
      this.connect()
      this.lockReconnect = false
    }, 1000 * this.reconnectCount) //重連次數(shù)越多，延時(shí)越久
  }

  // 重置心跳定時(shí)器/服務(wù)超時(shí)定時(shí)器
  private reset() {
    this.heartTimer && clearTimeout(this.heartTimer)

    this.serverTimer && clearTimeout(this.serverTimer)
  }

  // WebSocket連接狀態(tài)
  onReady() {
    return this.ws && this.ws.readyState
  }
}

export default SocketConnect

以上就是WebSocket中心跳檢測(cè)與斷開重連機(jī)制詳解的詳細(xì)內(nèi)容，更多關(guān)于WebSocket心跳檢測(cè)與斷開重連的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: