关键词搜索

源码搜索 ×
×

JavaScript WebSocket实现长连接通信连接重连

发布2018-11-22浏览7454次

详情内容

JavaScript-JS依靠WebSoket也可以像其他语言一样能够实现通信代码，不过使用的时候需要注意检查连接断开的情况。最基本的必须保证IP和端口都可以访问。

目录

WebSocket代码及测试

调用初始化代码

通信示例代码

测试效果

WebSocket连接关闭重连处理

连接断开问题

代码实现重连

重连效果

WebSocket代码及测试

调用初始化代码

  1.  CvNetVideo.Websocket = new WebSocket('ws://' + websocetHost + ":" + websocketPort);
  2.  CvNetVideo.WebsocketCallbackLoader = new WebsocketCallbackLoader();
  3.  CvNetVideo.WebsocketCallbackLoader.connect(CvNetVideo.Websocket);

通信示例代码

  1. /*
  2.  * WebsocketCallback Loader
  3. */
  4. import JT1078 from '../JX/JTT1078';
  5.  
  6. class WebsocketCallbackLoader {
  7.  
  8.     constructor() {
  9.         this.websocket = undefined;
  10.         this.onCallback = undefined;
  11.         this.sendMessage = this.sendMessage;
  12.         this.medisPackageAnalysis = new JT1078.MedisPackageAnalysis();
  13.         this.splitData = new JT1078.SplitData();
  14.         this.checkHelper = new JT1078.CheckHelper();
  15.         //分包缓存数据 按SIM做key
  16.         this.AllSimPackData = {};
  17.     }
  18.  
  19.     connect(websocket) {
  20.         this.websocket = websocket;
  21.         this.websocket.onopen = this.openSocket.bind(this);
  22.         this.websocket.onclose = this.closeSocket.bind(this);
  23.         this.websocket.binaryType = 'arraybuffer';
  24.         this.websocket.onmessage = this.receiveSocketMessage.bind(this);
  25.     }
  26.  
  27.     closeSocket(e) {
  28.         console.log('Websocket808 Disconnected!');
  29.     }
  30.  
  31.     openSocket() {
  32.         console.log('Websocket808 Open!');
  33.     }
  34.  
  35.     receiveSocketMessage(event) {
  36.         let data = new Uint8Array(event.data);
  37.         this.splitData.SplitDataBy7E(data, 18, data.byteLength - 18, (bGps) => {
  38.             if (bGps.length > 11 && this.checkHelper.CheckXOR(bGps, 0, bGps.length - 1) === bGps[bGps.length - 1])//效验通过
  39.             {
  40.                 this.JX808DataPD(new Uint8Array(bGps));
  41.             }
  42.         });
  43.     }
  44.     JX808DataPD(bGps) {
  45.         let head = new JT1078.JTHeader();
  46.         head.FillByBinary(bGps, 0);
  47.  
  48.         if (head.PackInfo) {
  49.             var sid = JT1078.GetFirstSerialNumber(head);
  50.             let AllPackData = AllSimPackData[head.Sim];
  51.             if (!AllPackData) {
  52.                 AllPackData = {};
  53.                 AllSimPackData[head.Sim] = AllPackData;
  54.             }
  55.  
  56.             var pd = AllPackData[sid];
  57.             if (pd) {
  58.                 if (!pd.PackData[head.PackInfo.Index]) {
  59.                     pd.PackData.Count += 1;
  60.                 }
  61.                 pd.PackData[head.PackInfo.Index] = JT1078.GetMsbBody(head, bGps);
  62.                 pd.LastRTime = DateTime.Now;
  63.                 if (this.CheckPackByPD(pd))
  64.                     AllPackData[sid] = null;
  65.             }
  66.             else {
  67.                 pd = {};
  68.                 pd.PackData = {};
  69.                 pd.PackData[head.PackInfo.Index] = JT1078.GetMsbBody(head, bGps);
  70.                 pd.PackData.Count = 1;
  71.                 pd.Head = new JT1078.JTHeader(head.MsgId, head.Sub, head.DataEncryptType, head.MsgLen, head.Sim, sid)
  72.                 pd.Head.PackInfo = new JT1078.JTPackageInfo(1, head.PackInfo.Sum);
  73.                 pd.StartTime = pd.LastRTime = DateTime.Now;
  74.                 AllPackData[sid] = pd;
  75.             }
  76.  
  77.         } else {
  78.             this.JX808Data(head, JT1078.GetMsbBody(head, bGps));
  79.         }
  80.     }
  81.     CheckPackByPD(pd) {
  82.         if (pd.PackData.Count === pd.Head.PackInfo.Sum)//判断数据包数和总包数是否相等
  83.         {
  84.             let tmpbuf = new Uint8Array();
  85.             let index = 0;
  86.             for (var i = 0; i < pd.Head.PackInfo.Sum; i++) {
  87.                 tmpbuf.set(pd.PackData[i], index);
  88.                 index += pd.PackData[i].byteLength;
  89.             }
  90.             //解析失败时返回true防止一直解析此错误数据
  91.             try {
  92.                 this.JX808Data(pd.Head, tmpbuf);
  93.             }
  94.             catch (ex)
  95.             {
  96.                 console.error(ex);
  97.                 return true;
  98.             }
  99.             return true;
  100.         }
  101.         return false;
  102.     }
  103.     JX808Data(head, bts) {
  104.         switch (head.MsgId) {
  105.  
  106.             case 0x1205://终端上传音视频资源列表数据格式
  107.                 var videolist = new JT1078.JTVideoListInfo();
  108.                 videolist.FillByBinary(bts, 0);
  109.                 this.onCallback(videolist);
  110.                 break;
  111.             case 0x1206://终端上传FTP完成通知
  112.                 break;
  113.             default:
  114.         }
  115.     }
  116.  
  117.     sendMessage(data, callback) {
  118.         this.onCallback = callback;
  119.         switch (this.websocket.readyState) {
  120.             case 0:
  121.                 console.log("Websocket808 readyState=CONNECTING");
  122.                 break;
  123.             case 1:
  124.                 this.websocket.send(data);
  125.                 break;
  126.             case 2:
  127.                 console.log("Websocket808 readyState=CLOSING");
  128.                 break;
  129.             case 3:
  130.                 console.log("Websocket808 readyState=CLOSED");
  131.                 break;
  132.  
  133.         }
  134.  
  135.     }
  136.  
  137. }
  138.  
  139. export default WebsocketCallbackLoader;

测试效果

WebSocket连接关闭重连处理

连接断开问题

代码实现重连

改造一下调用代码和发送消息代码：

  1.    static initWebsocket(websocetHost, websocketPort) {
  2.         CvNetVideo.WebsocketCallbackLoader = new WebsocketCallbackLoader(websocetHost, websocketPort);
  3.         CvNetVideo.WebsocketCallbackLoader.connect();
  4.     }
  1. /*
  2.  * WebsocketCallback Loader
  3. */
  4. import JT1078 from '../JX/JTT1078';
  5. import { setTimeout } from 'timers';
  6.  
  7. class WebsocketCallbackLoader {
  8.  
  9.     constructor(host, port) {
  10.         this.host = host;
  11.         this.port = port;
  12.         this.websocket = undefined;
  13.         this.onCallback = undefined;
  14.         this.sendMessage = this.sendMessage;
  15.         this.medisPackageAnalysis = new JT1078.MedisPackageAnalysis();
  16.         this.splitData = new JT1078.SplitData();
  17.         this.checkHelper = new JT1078.CheckHelper();
  18.         //分包缓存数据 按SIM做key
  19.         this.AllSimPackData = {};
  20.     }
  21.  
  22.     connect() {
  23.         var url = "ws://" + this.host + ":" + this.port;
  24.         console.log("Websocket808 WebSocket connect ..." + url);
  25.         this.websocket = new WebSocket(url);
  26.         this.websocket.onopen = this.openSocket.bind(this);
  27.         this.websocket.onclose = this.closeSocket.bind(this);
  28.         this.websocket.binaryType = 'arraybuffer';
  29.         this.websocket.onmessage = this.receiveSocketMessage.bind(this);
  30.     }
  31.  
  32.     closeSocket(e) {
  33.         console.log('Websocket808 Disconnected!');
  34.     }
  35.  
  36.     openSocket() {
  37.         console.log('Websocket808 Open!');
  38.     }
  39.  
  40.     receiveSocketMessage(event) {
  41.         let data = new Uint8Array(event.data);
  42.         this.splitData.SplitDataBy7E(data, 18, data.byteLength - 18, (bGps) => {
  43.             if (bGps.length > 11 && this.checkHelper.CheckXOR(bGps, 0, bGps.length - 1) === bGps[bGps.length - 1])//效验通过
  44.             {
  45.                 this.JX808DataPD(new Uint8Array(bGps));
  46.             }
  47.         });
  48.     }
  49.     JX808DataPD(bGps) {
  50.         let head = new JT1078.JTHeader();
  51.         head.FillByBinary(bGps, 0);
  52.  
  53.         if (head.PackInfo) {
  54.             var sid = JT1078.GetFirstSerialNumber(head);
  55.             let AllPackData = AllSimPackData[head.Sim];
  56.             if (!AllPackData) {
  57.                 AllPackData = {};
  58.                 AllSimPackData[head.Sim] = AllPackData;
  59.             }
  60.  
  61.             var pd = AllPackData[sid];
  62.             if (pd) {
  63.                 if (!pd.PackData[head.PackInfo.Index]) {
  64.                     pd.PackData.Count += 1;
  65.                 }
  66.                 pd.PackData[head.PackInfo.Index] = JT1078.GetMsbBody(head, bGps);
  67.                 pd.LastRTime = DateTime.Now;
  68.                 if (this.CheckPackByPD(pd))
  69.                     AllPackData[sid] = null;
  70.             }
  71.             else {
  72.                 pd = {};
  73.                 pd.PackData = {};
  74.                 pd.PackData[head.PackInfo.Index] = JT1078.GetMsbBody(head, bGps);
  75.                 pd.PackData.Count = 1;
  76.                 pd.Head = new JT1078.JTHeader(head.MsgId, head.Sub, head.DataEncryptType, head.MsgLen, head.Sim, sid)
  77.                 pd.Head.PackInfo = new JT1078.JTPackageInfo(1, head.PackInfo.Sum);
  78.                 pd.StartTime = pd.LastRTime = DateTime.Now;
  79.                 AllPackData[sid] = pd;
  80.             }
  81.  
  82.         } else {
  83.             this.JX808Data(head, JT1078.GetMsbBody(head, bGps));
  84.         }
  85.     }
  86.     CheckPackByPD(pd) {
  87.         if (pd.PackData.Count === pd.Head.PackInfo.Sum)//判断数据包数和总包数是否相等
  88.         {
  89.             let tmpbuf = new Uint8Array();
  90.             let index = 0;
  91.             for (var i = 0; i < pd.Head.PackInfo.Sum; i++) {
  92.                 tmpbuf.set(pd.PackData[i], index);
  93.                 index += pd.PackData[i].byteLength;
  94.             }
  95.             //解析失败时返回true防止一直解析此错误数据
  96.             try {
  97.                 this.JX808Data(pd.Head, tmpbuf);
  98.             }
  99.             catch (ex)
  100.             {
  101.                 console.error(ex);
  102.                 return true;
  103.             }
  104.             return true;
  105.         }
  106.         return false;
  107.     }
  108.     JX808Data(head, bts) {
  109.         switch (head.MsgId) {
  110.  
  111.             case 0x1205://终端上传音视频资源列表数据格式
  112.                 var videolist = new JT1078.JTVideoListInfo();
  113.                 videolist.FillByBinary(bts, 0);
  114.                 this.onCallback(videolist);
  115.                 break;
  116.             case 0x1206://终端上传FTP完成通知
  117.                 break;
  118.             default:
  119.         }
  120.     }
  121.  
  122.     printState() {
  123.         switch (this.websocket.readyState) {
  124.             case 0:
  125.                 console.log("Websocket808 readyState=CONNECTING");
  126.                 break;
  127.             case 1:
  128.                 console.log("Websocket808 readyState=OPEN");
  129.                 break;
  130.             case 2:
  131.                 console.log("Websocket808 readyState=CLOSING");
  132.                 break;
  133.             case 3:
  134.                 console.log("Websocket808 readyState=CLOSED");
  135.                 break;
  136.         }
  137.     }
  138.  
  139.     sendData(data)
  140.     {
  141.         this.websocket.send(data);
  142.     }
  143.  
  144.     sendMessage(data, callback) {
  145.         this.onCallback = callback;
  146.         this.printState();
  147.         if (this.websocket.readyState === 0) {
  148.             setTimeout(() => {
  149.                 this.sendData(data);
  150.             }, 3000);
  151.         } else if (this.websocket.readyState === 2 || this.websocket.readyState === 3) {
  152.             this.connect();
  153.             setTimeout(() => {
  154.                 this.printState();
  155.                 this.sendData(data);
  156.             }, 3000);
  157.         } else {
  158.             this.sendData(data);
  159.         }
  160.  
  161.     }
  162.  
  163. }
  164.  
  165. export default WebsocketCallbackLoader;

重连效果

 下面是连接断开后重连效果：

注意：重连是异步调用的，使用setTimeout(func,ms) 调用，因为连接创建需要时间，直接连很有可能正在连接。

使用队列处理连接失败数据

  1. /*
  2.  * WebsocketCallback Loader
  3. */
  4. import JT1078 from '../JX/JTT1078';
  5. import Queue from '../utils/queue';
  6.  
  7. class WebsocketCallbackLoader{
  8.  
  9.     constructor(host, port) {
  10.         this.host = host;
  11.         this.port = port;
  12.         this.websocket = undefined;
  13.         this.medisPackageAnalysis = new JT1078.MedisPackageAnalysis();
  14.         this.splitData = new JT1078.SplitData();
  15.         this.checkHelper = new JT1078.CheckHelper();
  16.         //分包缓存数据 按SIM做key
  17.         this.AllSimPackData = {};
  18.         this.sendCache = new Queue();
  19.         this.onCallback = {};
  20.     }
  21.  
  22.     connect() {
  23.         var url = "ws://" + this.host + ":" + this.port;
  24.         this.websocket = new WebSocket(url);
  25.         this.websocket.onopen = this.openSocket.bind(this);
  26.         this.websocket.onclose = this.closeSocket.bind(this);
  27.         this.websocket.binaryType = 'arraybuffer';
  28.         this.websocket.onmessage = this.receiveSocketMessage.bind(this);
  29.     }
  30.  
  31.     closeSocket(e) {
  32.         console.log('Websocket808 Disconnected!');
  33.     }
  34.  
  35.     openSocket() {
  36.         console.log('Websocket808 Open!');
  37.         while (this.sendCache.size() > 0) {
  38.             this.sendData(this.sendCache.dequeue());
  39.         }
  40.     }
  41.  
  42.     receiveSocketMessage(event) {
  43.         let data = new Uint8Array(event.data);
  44.         this.splitData.SplitDataBy7E(data, 18, data.byteLength - 18, (bGps) => {
  45.             if (bGps.length > 11 && this.checkHelper.CheckXOR(bGps, 0, bGps.length - 1) === bGps[bGps.length - 1])//效验通过
  46.             {
  47.                 this.JX808DataPD(new Uint8Array(bGps));
  48.             }
  49.         });
  50.     }
  51.     JX808DataPD(bGps) {
  52.         let head = new JT1078.JTHeader();
  53.         head.FillByBinary(bGps, 0);
  54.  
  55.         if (head.PackInfo) {
  56.             var sid = JT1078.GetFirstSerialNumber(head);
  57.             let AllPackData = this.AllSimPackData[head.Sim];
  58.             if (!AllPackData) {
  59.                 AllPackData = {};
  60.                 this.AllSimPackData[head.Sim] = AllPackData;
  61.             }
  62.  
  63.             var pd = AllPackData[sid];
  64.             if (pd) {  
  65.                 if (!pd.PackData[head.PackInfo.Index]) {
  66.                     pd.PackData.Count += 1;
  67.                     pd.PackData.totalLength += head.MsgLen;
  68.                 }
  69.                 pd.PackData[head.PackInfo.Index] = JT1078.GetMsbBody(head, bGps);
  70.                 pd.LastRTime = new Date();
  71.                 if (this.CheckPackByPD(pd))
  72.                     AllPackData[sid] = null;
  73.             }
  74.             else {
  75.                 pd = {};
  76.                 pd.PackData = {};
  77.                 pd.PackData[head.PackInfo.Index] = JT1078.GetMsbBody(head, bGps);
  78.                 pd.PackData.Count = 1;
  79.                 pd.PackData.totalLength = head.MsgLen;
  80.                 pd.Head = new JT1078.JTHeader(head.MsgId, head.Sub, head.DataEncryptType, head.MsgLen, head.Sim, sid);
  81.                 pd.Head.PackInfo = new JT1078.JTPackageInfo(1, head.PackInfo.Sum);
  82.                 pd.StartTime = pd.LastRTime = new Date();
  83.                 AllPackData[sid] = pd;
  84.             }
  85.  
  86.         } else {
  87.             this.JX808Data(head, JT1078.GetMsbBody(head, bGps));
  88.         }
  89.     }
  90.     CheckPackByPD(pd) {
  91.         if (pd.PackData.Count === pd.Head.PackInfo.Sum)//判断数据包数和总包数是否相等
  92.         {
  93.             let tmpbuf = new Uint8Array(pd.PackData.totalLength);
  94.             let index = 0;
  95.             for (var i = 1; i <= pd.Head.PackInfo.Sum; i++) {
  96.                 tmpbuf.set(pd.PackData[i], index);
  97.                 index += pd.PackData[i].byteLength;
  98.             }
  99.             //解析失败时返回true防止一直解析此错误数据
  100.             try {
  101.                 this.JX808Data(pd.Head, tmpbuf);
  102.             }
  103.             catch (ex) {
  104.                 console.error(ex);
  105.                 return true;
  106.             }
  107.             return true;
  108.         }
  109.         return false;
  110.     }
  111.     JX808Data(head, bts) {
  112.         switch (head.MsgId) {
  113.  
  114.             case 0x1205://终端上传音视频资源列表数据格式
  115.                 var videolist = new JT1078.JTVideoListInfo();
  116.                 videolist.FillByBinary(bts, 0);
  117.                 this.onCallback[0x1205](videolist);
  118.                 break;
  119.             case 0x1206://终端上传FTP完成通知
  120.                 var fileUploadEndInform = new JT1078.JTVideoFileUploadEndInform();
  121.                 fileUploadEndInform.FillByBinary(bts, 0);
  122.                 this.onCallback[0x1206](fileUploadEndInform);
  123.                 break;
  124.             default:
  125.         }
  126.     }
  127.  
  128.     printState() {
  129.         switch (this.websocket.readyState) {
  130.             case 0:
  131.                 console.log("Websocket808 readyState=CONNECTING");
  132.                 break;
  133.             case 1:
  134.                 console.log("Websocket808 readyState=OPEN");
  135.                 break;
  136.             case 2:
  137.                 console.log("Websocket808 readyState=CLOSING");
  138.                 break;
  139.             case 3:
  140.                 console.log("Websocket808 readyState=CLOSED");
  141.                 break;
  142.         }
  143.     }
  144.  
  145.     sendData(data)
  146.     {
  147.         this.websocket.send(data);
  148.     }
  149.  
  150.     sendMessage(data, callback, id) {
  151.  
  152.         this.onCallback[id] = callback;
  153.         if (this.websocket.readyState === 0) {
  154.             this.sendCache.enqueue(data);
  155.         } else if (this.websocket.readyState === 2 || this.websocket.readyState === 3) {
  156.             this.printState();
  157.             this.sendCache.enqueue(data);
  158.             this.connect();
  159.         } else {
  160.             this.sendData(data);
  161.         }
  162.  
  163.     }
  164.  
  165. }
  166.  
  167. export default WebsocketCallbackLoader;

 

相关技术文章

最新源码

下载排行榜
点击QQ咨询
开通会员
返回顶部
×
微信扫码支付 元
微信扫码支付
确定支付下载
请使用微信描二维码支付
×

提示信息

×

选择支付方式

  • 微信支付
  • 支付宝付款
确定支付下载