Netty websocket 推送数据压缩以 js解压

       在项目开发的时候利用基于Netty 的websocket项目，有时会发现在推送过程中经常不推送了。经过研究调查发现服务器在高并发的情况下，推送的数据流量接近带宽流量峰值，会导致带宽不足无法继续推送新的数据。

      为了解决这个问题：方法1：加大带宽。（花费多点钱买带宽流量）

                                     方法2：压缩数据。（减少网络传输带宽流量）

方法1没什么好说的，给钱就可以了。

主要讲讲方法2：压缩传输数据。（我网上我搜了好久没有比较完整的处理方法和代码，特记录一下处理过程给有需要的人参考）

首先：在websocket 服务端推送数据的时候，对要传输的数据进行压缩，这里我用GZIP进行压缩并用BASE64进行编码。代码如下：

 /**
     * 发送消息
     * @param set 频道通道集合
     * @param msg 发送消息内容
     */
    public static void sendBinaryMsg(CopyOnWriteArraySet<Channel> set,String msg,String channelName){
     
     if(set==null)return;
  try {
   if(set.size()==0)return;
   String frameType = "";
   msg = ZipUtil.gzip(msg);//压缩数据
   AttributeKey<String> key = AttributeKey.valueOf(channelName+"_binary");
   Iterator<Channel> it=set.iterator();
   while(it.hasNext()){
    Channel channel=it.next();
    if(channel.isActive()){
      if(channel.attr(key)!=null){
        frameType = channel.attr(key).get();
      }
     if(frameType != null && frameType.equals("false")){
      channel.writeAndFlush(new TextWebSocketFrame(msg));
     }else{
      channel.writeAndFlush(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(msg.getBytes())));
     }
    }else{
     set.remove(channel);
    }
   }
  } catch (Exception e) {
   e.printStackTrace();
  }
    }

 

此处用到 ZipUtil.java的gzip压缩:

 

 package com.world.common;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import java.util.zip.ZipOutputStream;

/**
 * 解压缩字符串工具类
 * @author zhanglinbo 20160827
 *
 */
public class ZipUtil {

 /**
  *
  * 功能：使用gzip进行压缩，然后再用Base64进行编码
  * @param 待压缩字符串
  * @return 返回压缩后字符串
  * @author zhanglinbo  20160827
  */
 @SuppressWarnings("restriction")
 public static String gzip(String primStr) {
  if (primStr == null || primStr.length() == 0) {
   return primStr;
  }

  ByteArrayOutputStream out = new ByteArrayOutputStream();
  
  GZIPOutputStream gzip = null;
  try {
   gzip = new GZIPOutputStream(out);
   gzip.write(primStr.getBytes());
   
  } catch (IOException e) {
   e.printStackTrace();
  } finally {
   if (gzip != null) {
    try {
     gzip.close();
     
    } catch (IOException e) {
     e.printStackTrace();
    }
   }
  }

  return  new sun.misc.BASE64Encoder().encode(out.toByteArray());
 }

 /**
  *
  * <p>
  * Description:使用gzip进行解压缩
  * 先对压缩数据进行BASE64解码。再进行Gzip解压
  * </p>
  *
  * @param compressedStr 压缩字符串
  * @return 返回解压字符串
  */
 @SuppressWarnings("restriction")
 public static String gunzip(String compressedStr) {
  if (compressedStr == null) {
   return null;
  }

  ByteArrayOutputStream out = new ByteArrayOutputStream();
  ByteArrayInputStream in = null;
  GZIPInputStream ginzip = null;
  byte[] compressed = null;
  String decompressed = null;
  try {
   compressed = new sun.misc.BASE64Decoder().decodeBuffer(compressedStr);
   in = new ByteArrayInputStream(compressed);
   ginzip = new GZIPInputStream(in);

   byte[] buffer = new byte[1024];
   int offset = -1;
   while ((offset = ginzip.read(buffer)) != -1) {
    out.write(buffer, 0, offset);
   }
   decompressed = out.toString();
  } catch (IOException e) {
   e.printStackTrace();
  } finally {
   if (ginzip != null) {
    try {
     ginzip.close();
    } catch (IOException e) {
    }
   }
   if (in != null) {
    try {
     in.close();
    } catch (IOException e) {
    }
   }
   if (out != null) {
    try {
     out.close();
    } catch (IOException e) {
    }
   }
  }

  return decompressed;
 }

 /**
  * 使用zip进行压缩
  *
  * @param str
  *            压缩前的文本
  * @return 返回压缩后的文本
  */
 @SuppressWarnings("restriction")
 public static final String zip(String str) {
  if (str == null)
   return null;
  byte[] compressed;
  ByteArrayOutputStream out = null;
  ZipOutputStream zout = null;
  String compressedStr = null;
  try {
   out = new ByteArrayOutputStream();
   zout = new ZipOutputStream(out);
   zout.putNextEntry(new ZipEntry("0"));
   zout.write(str.getBytes());
   zout.closeEntry();
   compressed = out.toByteArray();
   compressedStr = new sun.misc.BASE64Encoder().encodeBuffer(compressed);
  } catch (IOException e) {
   compressed = null;
  } finally {
   if (zout != null) {
    try {
     zout.close();
    } catch (IOException e) {
    }
   }
   if (out != null) {
    try {
     out.close();
    } catch (IOException e) {
    }
   }
  }
  return compressedStr;
 }

 /**
  * 使用zip进行解压缩
  *
  * @param compressed
  *            压缩后的文本
  * @return 解压后的字符串
  */
 @SuppressWarnings("restriction")
 public static final String unzip(String compressedStr) {
  if (compressedStr == null) {
   return null;
  }

  ByteArrayOutputStream out = null;
  ByteArrayInputStream in = null;
  ZipInputStream zin = null;
  String decompressed = null;
  try {
   byte[] compressed = new sun.misc.BASE64Decoder().decodeBuffer(compressedStr);
   out = new ByteArrayOutputStream();
   in = new ByteArrayInputStream(compressed);
   zin = new ZipInputStream(in);
   zin.getNextEntry();
   byte[] buffer = new byte[1024];
   int offset = -1;
   while ((offset = zin.read(buffer)) != -1) {
    out.write(buffer, 0, offset);
   }
   decompressed = out.toString();
  } catch (IOException e) {
   decompressed = null;
  } finally {
   if (zin != null) {
    try {
     zin.close();
    } catch (IOException e) {
    }
   }
   if (in != null) {
    try {
     in.close();
    } catch (IOException e) {
    }
   }
   if (out != null) {
    try {
     out.close();
    } catch (IOException e) {
    }
   }
  }
  return decompressed;
 }
 
}

 

--------------------------以上是netty websocket 服务端发送数据前的数据压缩----------------------------------

在浏览器端，利用js对压缩的数据进行解压缩。这里用到了pako项目的高效js解压缩组件。项目地址为：https://github.com/nodeca/pako 实测解压速度是毫秒级。

以下是测试例子，对压缩编码过的数据进行解压，打印日志，看是否正常

<!DOCTYPE html>
<html>
<meta charset="UTF-8">
<script src="https://cdn.jsdelivr.net/pako/1.0.3/pako.min.js"></script>
<script>
'use strict';

var pako = window.pako;

// In browser

// Get some base64 encoded binary data from the server. Imagine we got this:
var b64Data     = 'H4sIAAAAAAAAAJWSwW6DMAyGX4WbpQpFcRJM4Lqde+l6mCoOjOZQbcAE6QFVffcFWBq6VkjL4ZftfPplO7nA0fQQ5RFAFEdw6nfnakxtdzbRWDmWtpyAC+x3ry/bd8hhs4EYqraxXVnZfXOyvjbCkB8OqFKUkgRxd2LkDPGJCsaLeGKVJtKrLGdcBDhVHn4QzgKX4noDjl+4plL9g87+Dkd3YcIE6QBL4WF3J5hIlqFUzln9wgnyAHN8ovK2NiJUtMoue9YktVrQkvul+ViwLKNAZ7P3fDdrdhPF0Bu76dTccnbPTILuYZlM9ALWD753ynSAJapVOOHj8qgoYqjL7tPYbVkb/x3rtjHD2/A9FqpmcJV+qD/aL5caW8H1+gMYWAQw/gIAAA==';

// Decode base64 (convert ascii to binary)
var strData     = atob(b64Data);

// Convert binary string to character-number array
var charData    = strData.split('').map(function(x){return x.charCodeAt(0);});

// Turn number array into byte-array
var binData     = new Uint8Array(charData);

// Pako magic
var data        = pako.inflate(binData);

// Convert gunzipped byteArray back to ascii string:
var strData     = String.fromCharCode.apply(null, new Uint16Array(data));

// Output to console
console.log(strData);

</script>
</html>
<body>
Sending objects to server, run server code to see result.
</body>

 在项目中实际应用：

在监听websocket 消息方法中进行对数据解压：此处还判断了服务端返回的是二进制数据还是文本数据。

通过  ungzip 方法进行解压，并在回调函数中进行业务逻辑的后续处理。之所以要用回调函数，因为在等解压完成后，才能进行后续的操作，否则返回的数据可能是空的。

 //websocket 接收消息处理

    $this.socket.onmessage = function(result) {

    var json = null;

    if (result.data instanceof Blob) {    

        var blob = result.data;            

        //js中的blob没有没有直接读出其数据的方法，通过FileReader来读取相关数据      

        var reader = new FileReader();            

        reader.readAsText(blob);          

        //  当读取操作成功完成时调用.        

        reader.onload = function(evt){       

        if(evt.target.readyState == FileReader.DONE){ 

        var beford = lenght = evt.target.result.length;

        ungzip(evt.target.result,function(result){

                                                console.log("解压前："+beford+" 解压后："+result.length);

             

                                                if(result.indexOf("(")!=0){

          json = eval("("+result+")");

          }else{

          json = eval(result);

          }

         

          //处理业务逻辑

          $this.dealMessage(json);

        });

        

        }

        }

  }else{

   ungzip(result.data,function(result){

//console.log("解压后："+result);

      if(result.indexOf("(")!=0){

      json = eval("("+result+")");

      }else{

      json = eval(result);

      }

     //处理业务逻辑

     $this.dealMessage(json);

  });

 

      }

        };

 

 js 解压方法： 这里使用了seajs 的模块化处理引入pako.如不用seajs,直接在页面代码引入pako.min.js即可。

 

 /**

 * 解压缩字符串

 * @param zipData 经过 gzip压缩和base64编码的字符串

 * @param callback 回调函数 用解压缩后的数据进行处理后续操作

 * @author zhanglinbo 20160827

 */

function ungzip(zipData,callback){

var unZipData = "";//解压缩后数据

//引入pako模块进行数据的解压缩

seajs.use(["module_pako"],function(pako){

try{

// Decode base64 (convert ascii to binary)

var strData     = atob(zipData);

// Convert binary string to character-number array

var charData    = strData.split('').map(function(x){return x.charCodeAt(0);});

// Turn number array into byte-array

var binData     = new Uint8Array(charData);

// Pako magic

var data        = pako.inflate(binData);

// Convert gunzipped byteArray back to ascii string:

unZipData     = String.fromCharCode.apply(null, new Uint16Array(data));

}catch(e){

unZipData = zipData;//解压出现异常，说明数据未压缩。用原有数据进行操作

}

//利用回调进行处理后续操作

if($.isFunction(callback)){

callback(unZipData);

}

});

}

 

通过对数据的压缩传输和页面的解压，可以极大的剩下带宽。实测在数据长度大于2000以上的情况下压缩比例可以达到60-70%。对小数据字符串压缩比例不明显，有时反而比未压缩前大一点点。。不过影响不大。