Netty整合Protobuffer

时间 2019-11-16

原文原文链接

如今咱们都知道，rpc的三要素：IO模型，线程模型，而后就是数据交互模型，即咱们说的序列化和反序列化，如今咱们来看一下压缩比率最大的二进制序列化方式——Protobuffer,并且该方式是能够跨语言的，几乎大部分的语言均可以互相序列化和反序列化。javascript

要使用Protobuffer，须要先进行安装。由于本人使用的是mac，因此我使用的是.tar.gz的二进制压缩文件。html

其下载地址为https://github.com/protocolbuffers/protobuf/releases/tag/v3.5.0前端

下载完成后，依次执行如下命令java

tar -xzvf protobuf-all-3.5.0.tar.gznode

cd protobuf-3.5.0/jquery

./configuregit

makegithub

make checkweb

make installajax

安装须要的时间比较长，须要耐心等待，安装完成后，执行

admindeMacBook-Pro:protobuf-3.5.0 admin$ protoc --version
libprotoc 3.5.0

有显示红色字样，表示安装成功。

如今咱们来写一个hello world的样例

新建一个player.proto的文件

touch player.proto

vim player.proto

添入以下代码

option java_package = "com.guanjian.proto"; //此处为生成java文件的包名
option java_outer_classname = "PlayerModule"; //生成Java的类名

message PBPlayer{            //Protobuffer的类声明，会生成java的内部类
    required int64 playerId = 1; //required为必须字段，int64表示java的long类型，= 1这个1表示字段键名,每一个字段的键名不能重复
    required int32 age = 2; //int32表示java的int类型
    required string name = 3; //string表示java的String
    repeated int32 skills = 4; //repeated int32表示Java的List<Integer>
}

message PBResource{
required int64 gold = 1;
required int32 energy = 2;
}

保存后，输入命令

protoc ./player.proto --java_out=./

会在当前文件夹生成com/guanjian/proto的三个文件夹

cd com/guanjian/proto后，能够看到咱们生成的java文件

admindeMacBook-Pro:proto admin$ ls
PlayerModule.java

这里java文件里面的内容很是的多，也很是复杂。

如今咱们来看一下怎么来操做这个Java类。

在java中要使用protobuffer，须要添加依赖，版本号跟咱们安装的Protobuffer保持一致。

<dependency>
   <groupId>com.google.protobuf</groupId>
   <artifactId>protobuf-java</artifactId>
   <version>3.5.0</version>
</dependency>

public class PB2Bytes {
    public static void main(String[] args) {
        toBytes();
    }

    /**
     * 序列化
     */
    public static byte[] toBytes() {
        //获取一个PBPlayer的建造器（建造者模式）
        PlayerModule.PBPlayer.Builder builder = PlayerModule.PBPlayer.newBuilder();
        //建造一个PBPlayer对象
        PlayerModule.PBPlayer peter = builder.setPlayerId(101).setAge(21).setName("Peter").addSkills(1001)
                .build();
        //序列化成字节数组
        byte[] bytes = peter.toByteArray();
        System.out.println(Arrays.toString(bytes));
        return bytes;
    }
}

运行获得的结果为：

[8, 101, 16, 21, 26, 5, 80, 101, 116, 101, 114, 32, -23, 7]

以前咱们在.proto文件中设置了required标识，意思就是说在建造器中，咱们必须设置相应的属性，不然将会报错。

好比咱们把年龄取消掉，不设置年龄

public class PB2Bytes {
    public static void main(String[] args) {
        toBytes();
    }

    /**
     * 序列化
     */
    public static byte[] toBytes() {
        //获取一个PBPlayer的建造器（建造者模式）
        PlayerModule.PBPlayer.Builder builder = PlayerModule.PBPlayer.newBuilder();
        //建造一个PBPlayer对象
        PlayerModule.PBPlayer peter = builder.setPlayerId(101).setName("Peter").addSkills(1001)
                .build();
        //序列化成字节数组
        byte[] bytes = peter.toByteArray();
        System.out.println(Arrays.toString(bytes));
        return bytes;
    }
}

运行结果将抛出异常

如今咱们来增长反序列化

public class PB2Bytes {
    public static void main(String[] args) throws InvalidProtocolBufferException {
        byte[] bytes = toBytes();
        toPlayer(bytes);
    }

    /**
     * 序列化
     */
    public static byte[] toBytes() {
        //获取一个PBPlayer的建造器（建造者模式）
        PlayerModule.PBPlayer.Builder builder = PlayerModule.PBPlayer.newBuilder();
        //建造一个PBPlayer对象
        PlayerModule.PBPlayer peter = builder.setPlayerId(101).setAge(21).setName("Peter").addSkills(1001)
                .build();
        //序列化成字节数组
        byte[] bytes = peter.toByteArray();
        System.out.println(Arrays.toString(bytes));
        return bytes;
    }

    /**
     * 反序列化
     * @param bytes
     */
    public static void toPlayer(byte[] bytes) throws InvalidProtocolBufferException {
        PlayerModule.PBPlayer peter = PlayerModule.PBPlayer.parseFrom(bytes);
        System.out.println("playerId:" + peter.getPlayerId() + ",age:" + peter.getAge() +
        ",name:" + peter.getName() + ",skills:" + peter.getSkillsList().get(0));
    }
}

运行结果

[8, 101, 16, 21, 26, 5, 80, 101, 116, 101, 114, 32, -23, 7]
playerId:101,age:21,name:Peter,skills:1001

咱们仍是用Java自带的序列化和反序列化方式来对比一下Protobuffer的序列化和反序列化

首先，咱们定义一个相似PBPlayer的Java类Player

@Data
@Builder
public class Player implements Serializable {
    private long playerId;
    private int age;
    private String name;
    private List<Integer> skills;
}

这里咱们一样使用建造者模式来构建对象

public class Java2Bytes {
    public static void main(String[] args) throws IOException, ClassNotFoundException {
        byte[] bytes = toByes();
        toPlayer(bytes);
    }

    /**
     * 序列化
     * @return
     * @throws IOException
     */
    public static byte[] toByes() throws IOException {
        List<Integer> skills = new ArrayList<>(1);
        skills.add(1001);
        Player peter = Player.builder().playerId(101).age(21).name("Peter").skills(skills)
                .build();
        //字节数组流
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        //使用字节数组流来初始化一个对象流
        ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);
        //将Player对象写入对象流
        objectOutputStream.writeObject(peter);
        //获取字节数组
        byte[] bytes = byteArrayOutputStream.toByteArray();
        System.out.println(Arrays.toString(bytes));
        return bytes;
    }

    /**
     * 反序列化
     * @param bytes
     * @throws IOException
     * @throws ClassNotFoundException
     */
    public static void toPlayer(byte[] bytes) throws IOException, ClassNotFoundException {
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
        ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);
        Player peter = (Player) objectInputStream.readObject();
        System.out.println("playerId:" + peter.getPlayerId() + ",age:" + peter.getAge() +
                ",name:" + peter.getName() + ",skills:" + peter.getSkills().get(0));
    }
}

运行结果

[-84, -19, 0, 5, 115, 114, 0, 38, 99, 111, 109, 46, 103, 117, 97, 110, 106, 105, 97, 110, 46, 119, 101, 98, 115, 111, 99, 107, 101, 116, 46, 110, 101, 116, 116, 121, 46, 112, 98, 46, 80, 108, 97, 121, 101, 114, 119, -42, 102, 37, -49, -99, -17, 15, 2, 0, 4, 73, 0, 3, 97, 103, 101, 74, 0, 8, 112, 108, 97, 121, 101, 114, 73, 100, 76, 0, 4, 110, 97, 109, 101, 116, 0, 18, 76, 106, 97, 118, 97, 47, 108, 97, 110, 103, 47, 83, 116, 114, 105, 110, 103, 59, 76, 0, 6, 115, 107, 105, 108, 108, 115, 116, 0, 16, 76, 106, 97, 118, 97, 47, 117, 116, 105, 108, 47, 76, 105, 115, 116, 59, 120, 112, 0, 0, 0, 21, 0, 0, 0, 0, 0, 0, 0, 101, 116, 0, 5, 80, 101, 116, 101, 114, 115, 114, 0, 19, 106, 97, 118, 97, 46, 117, 116, 105, 108, 46, 65, 114, 114, 97, 121, 76, 105, 115, 116, 120, -127, -46, 29, -103, -57, 97, -99, 3, 0, 1, 73, 0, 4, 115, 105, 122, 101, 120, 112, 0, 0, 0, 1, 119, 4, 0, 0, 0, 1, 115, 114, 0, 17, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 73, 110, 116, 101, 103, 101, 114, 18, -30, -96, -92, -9, -127, -121, 56, 2, 0, 1, 73, 0, 5, 118, 97, 108, 117, 101, 120, 114, 0, 16, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 78, 117, 109, 98, 101, 114, -122, -84, -107, 29, 11, -108, -32, -117, 2, 0, 0, 120, 112, 0, 0, 3, -23, 120]
playerId:101,age:21,name:Peter,skills:1001

如此咱们能够看出ProtoBuffer的序列化结果字节数要小点多。

如今咱们来看一下Kryo的序列化出来的结果，要使用Kryo得修改一下Player类，由于Kryo反序列化必需要有无参构造器。首先放置依赖

<dependency>
   <groupId>com.esotericsoftware</groupId>
   <artifactId>kryo-shaded</artifactId>
   <version>4.0.2</version>
</dependency>

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Player implements Serializable {
    private long playerId;
    private int age;
    private String name;
    private List<Integer> skills;
}

此处咱们使用全参构造器来构造

public class Kryo2Bytes {
    public static void main(String[] args) {
        byte[] bytes = toByte();
        toPlayer(bytes);
    }

    /**
     * 序列化
     * @return
     */
    public static byte[] toByte() {
        List<Integer> skills = new ArrayList<>(1);
        skills.add(1001);
        Player peter = new Player(101,21,"Peter",skills);
        Kryo kryo = new Kryo();
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);
        kryo.writeObject(output,peter);
        output.close();
        byte[] bytes = byteArrayOutputStream.toByteArray();
        System.out.println(Arrays.toString(bytes));
        return bytes;
    }

    /**
     * 反序列化
     * @param bytes
     */
    public static void toPlayer(byte[] bytes) {
        Kryo kryo = new Kryo();
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
        Input input = new Input(byteArrayInputStream);
        Player peter = kryo.readObject(input, Player.class);
        System.out.println("playerId:" + peter.getPlayerId() + ",age:" + peter.getAge() +
                ",name:" + peter.getName() + ",skills:" + peter.getSkills().get(0));
        input.close();
    }
}

运行结果

[1, 42, 1, 80, 101, 116, 101, -14, -54, 1, 1, 0, 106, 97, 118, 97, 46, 117, 116, 105, 108, 46, 65, 114, 114, 97, 121, 76, 105, 115, -12, 1, 1, 2, -46, 15]
playerId:101,age:21,name:Peter,skills:1001

由以上结果可见，序列化字节数最少可能是ProtoBuffer,其次是Kryo，最长的是Java自己的序列化方式。

如今咱们来看一下JS的使用方法。

要使用JS的Protobuffer，须要先安装npm

brew install node

npm install google-protobuf

此时会在当前目录下生成一个node_modules,所有目录内容以下

而后使用protoc ./player.proto --js_out=library=myproto_libs,binary:. player.proto生成js版本的protobuffer文件

myproto_libs.js

将这些文件放入到项目的js目录下

而且注释掉google-protobuf.js掉最后一行代码

编辑咱们的index.html

<!DOCTYPE html>
<html>
	<head>
		<meta charset="utf-8" />
		<title></title>
	</head>
	<body>
		<script type="application/javascript" src="js/google-protobuf/google-protobuf.js"></script>
		<script type="application/javascript" src="js/myproto_libs.js"></script>
		<script type="application/javascript">
			var player = new proto.PBPlayer();
			player.setPlayerid(101);
			player.setAge(21);
			player.setName("Peter");
			player.addSkills(1001);
			var bytes = player.serializeBinary();
			var logs = "";
			for(var i=0;i < bytes.length;i++) {
				logs = logs + bytes[i] + ",";
			}
			console.log(logs);
			var peter = proto.PBPlayer.deserializeBinary(bytes);
			console.log("playerId:" + peter.getPlayerid() + ",age:" + peter.getAge() +
        ",name:" + peter.getName() + ",skills:" + peter.getSkillsList());
		</script>
	</body>
</html>

运行出来的效果以下

如今咱们来将Netty整合WebSocket 中的聊天改形成Protobuffer的传输，首先咱们须要定义知足业务需求的.proto文件

option java_package = "com.cloud.notification.chat.proto";
option java_outer_classname = "ChatModule";

message ChatMsg{
    optional string senderId = 1;
    optional string receiverId = 2;
    optional string msg = 3;
    optional string msgId = 4;
    optional int32 sign = 5;
    optional string createDate = 6;
}

message DataContent{
    required int32 action = 1;
    optional ChatMsg chatMsg = 2;
    optional string extand = 3;
}

按步骤生成咱们须要的java和js代码后，咱们来改造服务端。先将以前的ChatMsg类作一下修改。创建senderId，receiverId，msg的三参构造器就能够了。

@NoArgsConstructor
@RequiredArgsConstructor
@ToString
@Data
public class ChatMsg implements Serializable,Chat{
    @NonNull
    @JSONField(serializeUsing = ToStringSerializer.class)
    private Long senderId; //发送者的用户id
    @NonNull
    @JSONField(serializeUsing = ToStringSerializer.class)
    private Long receiverId; //接收者的用户id
    @NonNull
    private String msg;
    @JSONField(serializeUsing = ToStringSerializer.class)
    private Long msgId; //用于消息的签收
    private MsgSignFlagEnum signed; //消息签收状态
    private LocalDateTime createDate;

    @Override
    @Transactional
    public void save(Chat chatMsg) {
        ChatDao chatDao = SpringBootUtil.getBean(ChatDao.class);
        IdService idService = SpringBootUtil.getBean(IdService.class);
        ((ChatMsg)chatMsg).setMsgId(idService.genId());
        ((ChatMsg)chatMsg).setCreateDate(LocalDateTime.now());
        chatDao.saveChat((ChatMsg) chatMsg);
    }

    @Transactional
    @Override
    public void updateMsgSigned(List<Long> msgIdList) {
        ChatDao chatDao = SpringBootUtil.getBean(ChatDao.class);
        chatDao.updateMsgSigned(msgIdList);
    }

    @Transactional
    @Override
    public List<ChatMsg> findUnReadChat(Long acceptUserId) {
        ChatDao chatDao = SpringBootUtil.getBean(ChatDao.class);
        return chatDao.findUnReadMsg(acceptUserId);
    }
}

首先把文本传输为主的TextWebSocketFrame改成二进制传输为主的BinaryWebSocketFrame

/**
 * BinaryWebSocketFrame: 在netty中，用于为websocket专门处理二进制的对象，frame是消息的载体
 */
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<BinaryWebSocketFrame> {
    //用于记录和管理全部客户端的channel
    private static ChannelGroup users = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private Chat chatMsgService = ChatMsgFactory.createChatMsgService();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, BinaryWebSocketFrame msg) throws Exception {
        //获取客户端传输过来的消息
//        String content = msg.text();
        ByteBuf content = msg.content();
        log.info("content为:" + content.toString());
        Channel currentChannel = ctx.channel();
        //解析传输过来的消息转成聊天对象
        content.readerIndex(0);
        byte[] bytes = new byte[content.readableBytes()];
        content.readBytes(bytes);
//        DataContent dataContent = JSONObject.parseObject(content,DataContent.class);
        ChatModule.DataContent dataContent = ChatModule.DataContent.parseFrom(bytes);
        //获取聊天对象的动做
        Integer action = dataContent.getAction();

        if (action == MsgActionEnum.CONNECT.type) {
            //当websocket第一次open的时候，初始化channel,把用的channel和userId关联起来
            Long senderId = Long.parseLong(dataContent.getChatMsg().getSenderId());
            UserChannelRel.put(senderId,currentChannel);
            //测试
            users.stream().forEach(channel -> log.info(channel.id().asLongText()));
            UserChannelRel.output();
        }else if (action == MsgActionEnum.CHAT.type) {
            //聊天类型的消息，把聊天记录保存到数据库，同时标记消息的签收状态[未签收]
            ChatModule.ChatMsg chatMsg = dataContent.getChatMsg();
            String msgText = chatMsg.getMsg();
            Long receiverId = Long.parseLong(chatMsg.getReceiverId());
            Long senderId = Long.parseLong(chatMsg.getSenderId());
            //保存数据库
            ChatMsg javaChatMsg = new ChatMsg(Long.parseLong(chatMsg.getSenderId()),Long.parseLong(chatMsg.getReceiverId()),
                    chatMsg.getMsg());
            chatMsgService.save(javaChatMsg);
            ChatModule.ChatMsg.Builder builder = ChatModule.ChatMsg.newBuilder(chatMsg);
            log.info(javaChatMsg.getMsgId() + ",java的");
            ChatModule.ChatMsg newMsg = builder.setMsgId(String.valueOf(javaChatMsg.getMsgId())).build();
            log.info(newMsg.getMsgId() + ",protobuffer的");
            byte[] sendBytes = newMsg.toByteArray();
            ByteBuf buf = Unpooled.copiedBuffer(sendBytes);
            Channel receiverChannel = UserChannelRel.get(receiverId);
            if (receiverChannel == null) {
                //接收方离线状态，此处无需处理
            }else {
                Channel findChannel = users.find(receiverChannel.id());
                if (findChannel != null) {
                    findChannel.writeAndFlush(new BinaryWebSocketFrame(buf));
                }else {
                    //接收方离线，此处无需处理
                }
            }
        }else if (action == MsgActionEnum.SIGNED.type) {
            //签收消息类型，针对具体的消息进行签收，修改数据库中对应消息的签收状态[已签收]
            //扩展字段在signed类型的消息中，表明须要去签收的消息id,逗号间隔
            String msgIdsStr = dataContent.getExtand();
            log.info("extand为:" + msgIdsStr);
            String[] msgIds = msgIdsStr.split(",");
            List<Long> msgIdList = new ArrayList<>();
            for (String mId : msgIds) {
                if (!StringUtils.isEmpty(mId)) {
                    msgIdList.add(Long.valueOf(mId));
                }
            }
            log.info(msgIdList.toString());
            if (!CollectionUtils.isEmpty(msgIdList)) {
                //批量签收
                chatMsgService.updateMsgSigned(msgIdList);
            }
        }else if (action == MsgActionEnum.KEEPALIVE.type) {
            //心跳类型的消息
            log.info("收到来自channel为[" + currentChannel + "]的心跳包");
        }
    }

    /**
     * 当客户端链接服务端以后(打开链接)
     * 获取客户端的channel,而且放到ChannelGroup中去进行管理
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        users.add(ctx.channel());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        //当触发handlerRemoved,ChannelGroup会自动移除对应的客户端的channel
        //因此下面这条语句可不写
//        clients.remove(ctx.channel());
        log.info("客户端断开，channel对应的长id为:" + ctx.channel().id().asLongText());
        log.info("客户端断开，channel对应的短id为:" + ctx.channel().id().asShortText());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.channel().close();
        users.remove(ctx.channel());
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //IdleStateEvent是一个用户事件，包含读空闲/写空闲/读写空闲
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                log.info("进入读空闲");
            }else if (event.state() == IdleState.WRITER_IDLE) {
                log.info("进入写空闲");
            }else if (event.state() == IdleState.ALL_IDLE) {
                log.info("channel关闭前，用户数量为:" + users.size());
                //关闭无用的channel，以防资源浪费
                ctx.channel().close();
                log.info("channel关闭后，用户数量为:" + users.size());
            }

        }
    }
}

而后改造咱们的前端代码，这里比较重要的是前端的websocket默认是文本方式传输改为二进制方式传输CHAT.socket.binaryType = 'arraybuffer';

用户id为1的代码以下，这里须要注意，咱们用的各类id都是字符串类型的

<!DOCTYPE html>
<html>
	<head>
		<meta charset="utf-8" />
		<title></title>
	</head>
	<body>
		<div>发送消息:</div>
		<input type="text" id="msgContent" />
		<input type="button" value="发送" onclick="CHAT.chat('1','2',msgContent.value,app.CHAT,null)" />
		<input type="file" id="file" name="file">
		<input type="button" id="button" value="发送图片" >
		<div>接受消息:</div>
		<div id="receiveMsg" style="background-color: gainsboro;"></div>
		<script type="application/javascript" src="js/app.js"></script>
		<script type="application/javascript" src="js/mui.min.js"></script>
		<script type="application/javascript" src="js/jquery-3.3.1.min.js"></script>
		<script type="application/javascript" src="js/google-protobuf/google-protobuf.js"></script>
		<script type="application/javascript" src="js/myproto_libs.js"></script>
		<script type="application/javascript">
			window.CHAT = {
				socket: null,
				init: function() {
					if (window.WebSocket) {
						CHAT.socket = new WebSocket("ws://127.0.0.1:9999/ws");
						CHAT.socket.binaryType = 'arraybuffer';
						CHAT.socket.onopen = function() {
							console.log("链接创建成功");
							
							CHAT.chat("1",null,null,app.CONNECT,null);
							//每次链接的时候获取未读消息
							fetchUnReadMsg();
							//定时发送心跳,30秒一次
							setInterval("CHAT.keepalive()",30000);
						},
						CHAT.socket.onclose = function() {
							console.log("链接关闭");
						},
						CHAT.socket.onerror = function() {
							console.log("发生错误");
						},
						CHAT.socket.onmessage = function(e) {
							var receiveMsg = document.getElementById("receiveMsg");
							var html = receiveMsg.innerHTML;
							var chatMsg = proto.ChatMsg.deserializeBinary(e.data);
							console.log("senderId:" + chatMsg.getSenderid() + ",receiverId:" + chatMsg.getReceiverid() + ",msg:"
							+ chatMsg.getMsg() + ",MsgId:" + chatMsg.getMsgid());
							receiveMsg.innerHTML = html + "<br/>" + chatMsg.getMsg();
							//消息签收
							CHAT.chat(chatMsg.getReceiverid(),null,null,app.SIGNED,chatMsg.getMsgid());
						}
					}else {
						alert("浏览器不支持WebSocket协议...");
					}
				},
				chat: function(senderId,receiverId,msg,action,extand) {
					// var chatMsg = new app.ChatMsg(senderId,receiverId,msg,null);
					var chatMsg = new proto.ChatMsg();
					chatMsg.setSenderid(senderId);
					chatMsg.setReceiverid(receiverId);
					chatMsg.setMsg(msg);
					// var dataContent = new app.DataContent(action,chatMsg,extand);
					var dataContent = new proto.DataContent();
					dataContent.setAction(action);
					dataContent.setChatmsg(chatMsg);
					dataContent.setExtand(extand);
					var bytes = dataContent.serializeBinary();
					var intView = new Int8Array(bytes);
					CHAT.socket.send(intView);
				},
				keepalive: function() {
					CHAT.chat("1",null,null,app.KEEPALIVE,null);
					fetchUnReadMsg();
				}
			}
			CHAT.init();
			function fetchUnReadMsg() {
				mui.ajax('http://127.0.0.1:8008/notification-anon/getunreadmeg?receiverid=1',{
					data:{},
					dataType:'json',//服务器返回json格式数据
					type:'post',//HTTP请求类型
					timeout:10000,//超时时间设置为10秒；
					success:function(data){
						if (data.code == 200) {
							var contactList = data.data;
							var ids = "";
							console.log(JSON.stringify(contactList));
							var receiveMsg = document.getElementById("receiveMsg");
							for (var i = 0;i < contactList.length;i++) {
								var msgObj = contactList[i];
								var html = receiveMsg.innerHTML;
								receiveMsg.innerHTML = html + "<br/>" + msgObj.msg;
								ids = ids + msgObj.msgId + ",";
							}
							//批量签收未读消息
							CHAT.chat("1",null,null,app.SIGNED,ids);
						}
					}
				});
			}
			$(function () {
			        $("#button").click(function () {
			            var form = new FormData();
			            form.append("file", document.getElementById("file").files[0]);
			             $.ajax({
			                 url: "http://xxx.xxx.xxx.xxx:8010/files-anon/fdfsupload",        //后台url
			                 data: form,
			                 cache: false,
			                 async: false,
			                 type: "POST",                   //类型，POST或者GET
			                 dataType: 'json',              //数据返回类型，能够是xml、json等
			                 processData: false,
			                 contentType: false,
			                 success: function (data) {      //成功，回调函数
			                     if (data.code == 200) {
			                     	console.log(data.data);
						CHAT.chat("1","2","<img src='" + data.data + "' height='200' width='200' />",app.CHAT,null);
			                     }   
			                 }
			             });
			
			        })
			
			    })
		</script>
	</body>
</html>

用户id为2的代码以下

<!DOCTYPE html>
<html>
	<head>
		<meta charset="utf-8" />
		<title></title>
	</head>
	<body>
		<div>发送消息:</div>
		<input type="text" id="msgContent" />
		<input type="button" value="发送" onclick="CHAT.chat('2','1',msgContent.value,app.CHAT,null)" />
		<input type="file" id="file" name="file">
		<input type="button" id="button" value="发送图片" >
		<div>接受消息:</div>
		<div id="receiveMsg" style="background-color: gainsboro;"></div>
		<script type="application/javascript" src="js/app.js"></script>
		<script type="application/javascript" src="js/mui.min.js"></script>
		<script type="application/javascript" src="js/jquery-3.3.1.min.js"></script>
		<script type="application/javascript" src="js/google-protobuf/google-protobuf.js"></script>
		<script type="application/javascript" src="js/myproto_libs.js"></script>
		<script type="application/javascript">
			window.CHAT = {
				socket: null,
				init: function() {
					if (window.WebSocket) {
						CHAT.socket = new WebSocket("ws://127.0.0.1:9999/ws");
						CHAT.socket.binaryType = 'arraybuffer';
						CHAT.socket.onopen = function() {
							console.log("链接创建成功");
							
							CHAT.chat("2",null,null,app.CONNECT,null);
							//每次链接的时候获取未读消息
							fetchUnReadMsg();
							//定时发送心跳,30秒一次
							setInterval("CHAT.keepalive()",30000);
						},
						CHAT.socket.onclose = function() {
							console.log("链接关闭");
						},
						CHAT.socket.onerror = function() {
							console.log("发生错误");
						},
						CHAT.socket.onmessage = function(e) {
							var receiveMsg = document.getElementById("receiveMsg");
							var html = receiveMsg.innerHTML;
							var chatMsg = proto.ChatMsg.deserializeBinary(e.data);
							console.log("senderId:" + chatMsg.getSenderid() + ",receiverId:" + chatMsg.getReceiverid() + ",msg:"
							+ chatMsg.getMsg() + ",MsgId:" + chatMsg.getMsgid());
							receiveMsg.innerHTML = html + "<br/>" + chatMsg.getMsg();
							//消息签收
							CHAT.chat(chatMsg.getReceiverid(),null,null,app.SIGNED,chatMsg.getMsgid());
						}
					}else {
						alert("浏览器不支持WebSocket协议...");
					}
				},
				chat: function(senderId,receiverId,msg,action,extand) {
					// var chatMsg = new app.ChatMsg(senderId,receiverId,msg,null);
					var chatMsg = new proto.ChatMsg();
					chatMsg.setSenderid(senderId);
					chatMsg.setReceiverid(receiverId);
					chatMsg.setMsg(msg);
					// var dataContent = new app.DataContent(action,chatMsg,extand);
					var dataContent = new proto.DataContent();
					dataContent.setAction(action);
					dataContent.setChatmsg(chatMsg);
					dataContent.setExtand(extand);
					var bytes = dataContent.serializeBinary();
					var intView = new Int8Array(bytes);
					CHAT.socket.send(intView);
				},
				keepalive: function() {
					CHAT.chat("2",null,null,app.KEEPALIVE,null);
					fetchUnReadMsg();
				}
			}
			CHAT.init();
			function fetchUnReadMsg() {
				mui.ajax('http://127.0.0.1:8008/notification-anon/getunreadmeg?receiverid=2',{
					data:{},
					dataType:'json',//服务器返回json格式数据
					type:'post',//HTTP请求类型
					timeout:10000,//超时时间设置为10秒；
					success:function(data){
						if (data.code == 200) {
							var contactList = data.data;
							var ids = "";
							console.log(JSON.stringify(contactList));
							var receiveMsg = document.getElementById("receiveMsg");
							for (var i = 0;i < contactList.length;i++) {
								var msgObj = contactList[i];
								var html = receiveMsg.innerHTML;
								receiveMsg.innerHTML = html + "<br/>" + msgObj.msg;
								ids = ids + msgObj.msgId + ",";
							}
							//批量签收未读消息
							CHAT.chat("2",null,null,app.SIGNED,ids);
						}
					}
				});
			}
			$(function () {
			        $("#button").click(function () {
			            var form = new FormData();
			            form.append("file", document.getElementById("file").files[0]);
			             $.ajax({
			                 url: "http://xxx.xxx.xxx.xxx:8010/files-anon/fdfsupload",        //后台url
			                 data: form,
			                 cache: false,
			                 async: false,
			                 type: "POST",                   //类型，POST或者GET
			                 dataType: 'json',              //数据返回类型，能够是xml、json等
			                 processData: false,
			                 contentType: false,
			                 success: function (data) {      //成功，回调函数
			                     if (data.code == 200) {
			                     	console.log(data.data);
							CHAT.chat("2","1","<img src='" + data.data + "' height='200' width='200' />",app.CHAT,null);
			                     }   
			                 }
			             });
			
			        })
			
			    })
		</script>
	</body>
</html>