Netty整合WebSocket

时间  2019-11-13
标签 netty 整合 websocket 栏目 Netty 繁體版
原文   原文链接

WebSocket协议是基于 javascript

TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通讯——容许服务器主动发送信息给客户端 ，它是先进行一次Http的链接，链接成功后转为TCP链接。html

如今咱们来作一个WebSocket HelloWorld,意思为接收一条WebSocket客户端发送过来的消息，而后刷到全部链接上的客户端，你们均可以看到这条消息。前端

@Slf4j
@AllArgsConstructor
public class WebSocketServer {
    private int port;

    public void run() throws InterruptedException {
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss,worker)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .childOption(ChannelOption.TCP_NODELAY,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //web基于http协议的解码器
                            ch.pipeline().addLast(new HttpServerCodec());
                            //对大数据流的支持
                            ch.pipeline().addLast(new ChunkedWriteHandler());
                            //对http message进行聚合，聚合成FullHttpRequest或FullHttpResponse
                            ch.pipeline().addLast(new HttpObjectAggregator(1024 * 64));
                            //websocket服务器处理对协议，用于指定给客户端链接访问的路径
                            //该handler会帮你处理一些繁重的复杂的事
                            //会帮你处理握手动做：handshaking(close,ping,pong) ping + pong = 心跳
                            //对于websocket来说，都是以frames进行传输的，不一样的数据类型对应的frames也不一样
                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws"));
                            //添加咱们的自定义channel处理器
                            ch.pipeline().addLast(new WebSocketHandler());
                        }
                    });
            log.info("服务器启动中");
            ChannelFuture future = bootstrap.bind(port).sync();
            future.channel().closeFuture().sync();
        } finally {
            worker.shutdownGracefully();
            boss.shutdownGracefully();
        }

    }
}

channel处理器java

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

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        //获取客户端传输过来的消息
        String content = msg.text();
        log.info("接收到的数据" + content);
        clients.stream().forEach(channel -> channel.writeAndFlush(
                new TextWebSocketFrame("[服务器在]" + LocalDateTime.now() + "接收到消息:" + content)
        ));
        //下面的方法与上面一致
//        clients.writeAndFlush(new TextWebSocketFrame("[服务器在]" + LocalDateTime.now() +
//                "接收到消息:" + content));
    }

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

服务器启动jquery

@SpringBootApplication
public class WebsocketApplication {

   public static void main(String[] args) throws InterruptedException {
      SpringApplication.run(WebsocketApplication.class, args);
      new WebSocketServer(10101).run();
   }

}

客户端写一个html文件，代码以下web

<!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()" />
        
        <div>接受消息:</div>
        <div id="receiveMsg" style="background-color: gainsboro;"></div>
        
        <script type="application/javascript">
            window.CHAT = {
                socket: null,
                init: function() {
                    if (window.WebSocket) {
                        CHAT.socket = new WebSocket("ws://127.0.0.1:10101/ws");
                        CHAT.socket.onopen = function() {
                            console.log("链接创建成功");
                        },
                        CHAT.socket.onclose = function() {
                            console.log("链接关闭");
                        },
                        CHAT.socket.onerror = function() {
                            console.log("发生错误");
                        },
                        CHAT.socket.onmessage = function(e) {
                            console.log("接收到消息" + e.data);
                            var receiveMsg = document.getElementById("receiveMsg");
                            var html = receiveMsg.innerHTML;
                            receiveMsg.innerHTML = html + "<br/>" + e.data;
                        }
                    }else {
                        alert("浏览器不支持WebSocket协议...");
                    }
                },
                chat: function() {
                    var msg = document.getElementById("msgContent");
                    CHAT.socket.send(msg.value);
                }
            }
            CHAT.init();
        </script>
    </body>
</html>

打开浏览器以下ajax

此时咱们输入一段话，发送数据库

咱们能够看到全部打开的链接，都会收到相同的消息。json

此时服务器的日志bootstrap

2019-10-16 22:31:16.066  INFO 1376 --- [ntLoopGroup-3-5] c.g.w.netty.websocket.WebSocketHandler   : 接收到的数据helloworld
2019-10-16 22:31:33.131  INFO 1376 --- [ntLoopGroup-3-5] c.g.w.netty.websocket.WebSocketHandler   : 接收到的数据你好，中国

若是咱们关闭一个页面，服务器的日志为

2019-10-16 22:36:39.390  INFO 1376 --- [ntLoopGroup-3-7] c.g.w.netty.websocket.WebSocketHandler   : 客户端断开，channel对应的长id为:acde48fffe001122-00000560-00000007-9ca78ac7e2b907ab-0b15dfcb
2019-10-16 22:36:39.390  INFO 1376 --- [ntLoopGroup-3-7] c.g.w.netty.websocket.WebSocketHandler   : 客户端断开，channel对应的短id为:0b15dfcb

如今咱们来作一个点对点的聊天功能。

首先WebSocketServer添加空闲处理器

@Slf4j
@AllArgsConstructor
public class WebSocketServer {
    private int port;

    public void run() throws InterruptedException {
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss,worker)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .childOption(ChannelOption.TCP_NODELAY,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //web基于http协议的解码器
                            ch.pipeline().addLast(new HttpServerCodec());
                            //对大数据流的支持
                            ch.pipeline().addLast(new ChunkedWriteHandler());
                            //对http message进行聚合，聚合成FullHttpRequest或FullHttpResponse
                            ch.pipeline().addLast(new HttpObjectAggregator(1024 * 64));
                            //websocket服务器处理对协议，用于指定给客户端链接访问的路径
                            //该handler会帮你处理一些繁重的复杂的事
                            //会帮你处理握手动做：handshaking(close,ping,pong) ping + pong = 心跳
                            //对于websocket来说，都是以frames进行传输的，不一样的数据类型对应的frames也不一样
                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws"));
                            //针对客户端，若是1分钟时没有向服务端发送读写心跳(All),则主动断开
                            //若是是读空闲或者是写空闲，不处理
                            ch.pipeline().addLast(new IdleStateHandler(20,40,60));
                            //添加咱们的自定义channel处理器
                            ch.pipeline().addLast(new WebSocketHandler());
                        }
                    });
            log.info("服务器启动中");
            ChannelFuture future = bootstrap.bind(port).sync();
            future.channel().closeFuture().sync();
        } finally {
            worker.shutdownGracefully();
            boss.shutdownGracefully();
        }

    }
}

而后咱们须要增长两个实体类——消息类和聊天类

首先添加一个消息的接口

public interface Chat {
    /**
     * 保存消息到数据库
     * @param chatMsg
     */
    public void save(Chat chatMsg);

    /**
     * 签名消息
     * @param msgIdList
     */
    public void updateMsgSigned(List<Long> msgIdList);

    /**
     * 查找未读未签名消息
     * @param acceptUserId
     * @return
     */
    public List<ChatMsg> findUnReadChat(Long acceptUserId);
}

添加一个消息的类来实现该接口

@ToString
@Data
public class ChatMsg implements Serializable,Chat{
    @JSONField(serializeUsing = ToStringSerializer.class)
    private Long senderId; //发送者的用户id
    @JSONField(serializeUsing = ToStringSerializer.class)
    private Long receiverId; //接收者的用户id
    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);
    }
}

其中MsgSignFlagEnum为一个枚举类型，代码以下

public enum MsgSignFlagEnum implements LocalisableAll {
    unsign(0,"未签收"),
    signed(1,"已签收");

    public final int type;
    public final String value;

    private MsgSignFlagEnum(int type,String value) {
        this.type = type;
        this.value = value;
    }

    @Override
    public int getType() {
        return type;
    }

    @Override
    public String getValue() {
        return value;
    }
}

创建一个消息的工厂

public class ChatMsgFactory {
    public static Chat createChatMsgService() {
        return new ChatMsg();
    }
}

而后是聊天类

@Data
public class DataContent implements Serializable {
    private Integer action; //动做类型
    private ChatMsg chatMsg; //用户的聊天内容entity
    private String extand; //扩展字段
}

咱们根据动做类型来定义一个枚举

public enum MsgActionEnum {
    CONNECT(1,"第一次(或重连)初始化链接"),
    CHAT(2,"聊天消息"),
    SIGNED(3,"消息签收"),
    KEEPALIVE(4,"客户端保持心跳");

    public final Integer type;
    public final String content;

    private MsgActionEnum(Integer type,String content) {
        this.type = type;
        this.content = content;
    }
}

在写WebSocketHandler以前，咱们须要将用户Id跟Channel作一个绑定

/**
 * 用户id和channel的关联关系的处理
 */
@Slf4j
public class UserChannelRel {
    private static Map<Long,Channel> manager = new HashMap<>();

    public static void put(Long senderId,Channel channel) {
        manager.put(senderId,channel);
    }

    public static Channel get(Long senderId) {
        return manager.get(senderId);
    }

    public static void output() {
        manager.entrySet().stream().forEach(entry ->
            log.info("UserId:" + entry.getKey() + ",ChannelId:" +
                    entry.getValue().id().asLongText())
        );
    }
}

最后WebSocketHandler改写以下

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

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

        if (action == MsgActionEnum.CONNECT.type) {
            //当websocket第一次open的时候，初始化channel,把用的channel和userId关联起来
            Long senderId = dataContent.getChatMsg().getSenderId();
            UserChannelRel.put(senderId,currentChannel);
            //测试
            users.stream().forEach(channel -> log.info(channel.id().asLongText()));
            UserChannelRel.output();
        }else if (action == MsgActionEnum.CHAT.type) {
            //聊天类型的消息，把聊天记录保存到数据库，同时标记消息的签收状态[未签收]
            ChatMsg chatMsg = dataContent.getChatMsg();
            String msgText = chatMsg.getMsg();
            Long receiverId = chatMsg.getReceiverId();
            Long senderId = chatMsg.getSenderId();
            //保存数据库
            chatMsgService.save(chatMsg);
            Channel receiverChannel = UserChannelRel.get(receiverId);
            if (receiverChannel == null) {
                //接收方离线状态，此处无需处理
            }else {
                Channel findChannel = users.find(receiverChannel.id());
                if (findChannel != null) {
                    findChannel.writeAndFlush(new TextWebSocketFrame(
                            JSONObject.toJSONString(chatMsg)
                    ));
                }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());
            }

        }
    }
}

此处咱们能够看到有两个地方注释了//接收方离线状态，此处无需处理。因此咱们须要在用户上线的时候获取未签名的消息，只须要经过Controller从数据库获取就好

@RestController
public class UnReadMessageController {
   private Chat chatService = ChatMsgFactory.createChatMsgService();

   @SuppressWarnings("unchecked")
   @PostMapping("/notification-anon/getunreadmeg")
   public Result<List<ChatMsg>> getUnReadMessage(@RequestParam("receiverid") Long receiverId) {
      return Result.success(chatService.findUnReadChat(receiverId));
   }
}

因为咱们登陆用的是OAuth框架，在user模块添加

@PostMapping("/users-anon/finduser")
public LoginAppUser findUserByName(@RequestParam("username") String username) {
    return appUserService.findByUsername(username);
}

修改网关登陆，登陆后能够获取用户的信息，首先添加User模块的feign

@FeignClient("user-center")
public interface UserClient {
    @PostMapping("/users-anon/finduser")
    LoginAppUser findUserByName(@RequestParam("username") String username);
}
/**
 * 登录、刷新token、退出
 *
 * @author 关键
 */
@Slf4j
@RestController
public class TokenController {

    @Autowired
    private Oauth2Client oauth2Client;
    @Autowired
    private UserClient userClient;

    /**
     * 系统登录<br>
     * 根据用户名登陆<br>
     * 采用oauth2密码模式获取access_token和refresh_token
     *
     * @param username
     * @param password
     * @return
     */
    @SuppressWarnings("unchecked")
    @PostMapping("/sys/login")
    public Result<Map> login(@RequestParam String username,@RequestParam String password) {
        Map<String, String> parameters = new HashMap<>();
        parameters.put(OAuth2Utils.GRANT_TYPE, "password");
        parameters.put(OAuth2Utils.CLIENT_ID, "system");
        parameters.put("client_secret", "system");
        parameters.put(OAuth2Utils.SCOPE, "app");
//    parameters.put("username", username);
        // 为了支持多类型登陆，这里在username后拼装上登陆类型
        parameters.put("username", username + "|" + CredentialType.USERNAME.name());
        parameters.put("password", password);

        Map<String, Object> tokenInfo = oauth2Client.postAccessToken(parameters);
        AppUser user = userClient.findUserByName(username);
        tokenInfo.put("user",user);
        saveLoginLog(username, "用户名密码登录");

        return Result.success(tokenInfo);
    }

    @Autowired
    private LogClient logClient;

    /**
     * 登录日志
     *
     * @param username
     */
    private void saveLoginLog(String username, String remark) {
        log.info("{}登录", username);
        // 异步
        CompletableFuture.runAsync(() -> {
            try {
                Log log = Log.builder().username(username).module(LogModule.LOGIN).remark(remark).createTime(new Date())
                        .build();
                logClient.save(log);
            } catch (Exception e) {
                // do nothing
            }

        });
    }

    /**
     * 系统刷新refresh_token
     *
     * @param refresh_token
     * @return
     */
    @PostMapping("/sys/refresh_token")
    public Map<String, Object> refresh_token(String refresh_token) {
        Map<String, String> parameters = new HashMap<>();
        parameters.put(OAuth2Utils.GRANT_TYPE, "refresh_token");
        parameters.put(OAuth2Utils.CLIENT_ID, "system");
        parameters.put("client_secret", "system");
        parameters.put(OAuth2Utils.SCOPE, "app");
        parameters.put("refresh_token", refresh_token);

        return oauth2Client.postAccessToken(parameters);
    }

    /**
     * 退出
     *
     * @param access_token
     */
    @GetMapping("/sys/logout")
    public void logout(String access_token, @RequestHeader(required = false, value = "Authorization") String token) {
        if (StringUtils.isBlank(access_token)) {
            if (StringUtils.isNoneBlank(token)) {
                access_token = token.substring(OAuth2AccessToken.BEARER_TYPE.length() + 1);
            }
        }
        oauth2Client.removeToken(access_token);
    }
}

作如上修改后，咱们登陆后能够得到用户的Id,以上省略Dao,Mapper

这个时候服务端就所有完成了，咱们再来看一下客户端

先创建一个全局变量app.js

window.app = {
	
	/**
	 * 和后端的枚举对应
	 */
	CONNECT: 1,			//"第一次(或重连)初始化链接"),
	CHAT: 2,			//"聊天消息"),
	SIGNED: 3,			//"消息签收"),
	KEEPALIVE: 4,		//"客户端保持心跳");

	/**
	* 和后端的ChatMsg聊天模型对象保持一致
	* @param {Object} senderId
	* @param {Object} receiverId
	* @param {Object} msg
	* @param {Object} msgId
	*/
	ChatMsg: function(senderId,receiverId,msg,msgId) {
		this.senderId = senderId;
		this.receiverId = receiverId;
		this.msg = msg;
		this.msgId = msgId;
	},

	/**
	* 构建消息模型对象
	* @param {Object} action
	* @param {Object} chatMsg
	* @param {Object} extand
	*/
	DataContent: function(action,chatMsg,extand) {
		this.action = action;
		this.chatMsg = chatMsg;
		this.extand = extand;
	}
}

因为这里只是聊天样例，没有处理登陆功能，咱们如今就以1，2来表明两个用户的id

用户id为1的代码，文件名index.html

<!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)" />
		
		<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">
			window.CHAT = {
				socket: null,
				init: function() {
					if (window.WebSocket) {
						CHAT.socket = new WebSocket("ws://127.0.0.1:9999/ws");
						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) {
							console.log("接收到消息" + e.data);
							var receiveMsg = document.getElementById("receiveMsg");
							var html = receiveMsg.innerHTML;
							var chatMsg = JSON.parse(e.data);
							receiveMsg.innerHTML = html + "<br/>" + chatMsg.msg;
							//消息签收
							CHAT.chat(chatMsg.receiverId,null,null,app.SIGNED,String(chatMsg.msgId));
						}
					}else {
						alert("浏览器不支持WebSocket协议...");
					}
				},
				chat: function(senderId,receiverId,msg,action,extand) {
					var chatMsg = new app.ChatMsg(senderId,receiverId,msg,null);
					var dataContent = new app.DataContent(action,chatMsg,extand);
					CHAT.socket.send(JSON.stringify(dataContent));
				},
				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);
						}
					}
				});
			}
		</script>
	</body>
</html>

用户id为2的代码,文件名receive.html

<!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)" />
		
		<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">
			window.CHAT = {
				socket: null,
				init: function() {
					if (window.WebSocket) {
						CHAT.socket = new WebSocket("ws://127.0.0.1:9999/ws");
						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) {
							console.log("接收到消息" + e.data);
							var receiveMsg = document.getElementById("receiveMsg");
							var html = receiveMsg.innerHTML;
							var chatMsg = JSON.parse(e.data);
							receiveMsg.innerHTML = html + "<br/>" + chatMsg.msg;
							//消息签收
							CHAT.chat(chatMsg.receiverId,null,null,app.SIGNED,String(chatMsg.msgId));
						}
					}else {
						alert("浏览器不支持WebSocket协议...");
					}
				},
				chat: function(senderId,receiverId,msg,action,extand) {
					var chatMsg = new app.ChatMsg(senderId,receiverId,msg,null);
					var dataContent = new app.DataContent(action,chatMsg,extand);
					CHAT.socket.send(JSON.stringify(dataContent));
				},
				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);
						}
					}
				});
		</script>
	</body>
</html>

这里都是经过Json来作数据的序列化的，以后会修改成ProtoBuffer来处理。

如今来增长发图片的功能。首先咱们须要搭建好一个fastdfs服务器，具体能够参考分布式文件系统FastDFS安装配置 以及Springboot 2.0+FastDFS开发配置 。

这里只放入咱们须要的Controller

@RestController
public class FastDFSController {
    @Autowired
    private FastDFSClientWrapper dfsClient;

    @PostMapping("/files-anon/fdfsupload")
    @SuppressWarnings("unchecked")
    public Result<String> upload(@RequestParam("file") MultipartFile file) throws Exception {
        String imgUrl = dfsClient.uploadFile(file);
        return Result.success(imgUrl);
    }
}

而后修改咱们的用户id为1的前端代码

<!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">
			window.CHAT = {
				socket: null,
				init: function() {
					if (window.WebSocket) {
						CHAT.socket = new WebSocket("ws://127.0.0.1:9999/ws");
						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) {
							console.log("接收到消息" + e.data);
							var receiveMsg = document.getElementById("receiveMsg");
							var html = receiveMsg.innerHTML;
							var chatMsg = JSON.parse(e.data);
							receiveMsg.innerHTML = html + "<br/>" + chatMsg.msg;
							//消息签收
							CHAT.chat(chatMsg.receiverId,null,null,app.SIGNED,String(chatMsg.msgId));
						}
					}else {
						alert("浏览器不支持WebSocket协议...");
					}
				},
				chat: function(senderId,receiverId,msg,action,extand) {
					var chatMsg = new app.ChatMsg(senderId,receiverId,msg,null);
					var dataContent = new app.DataContent(action,chatMsg,extand);
					CHAT.socket.send(JSON.stringify(dataContent));
				},
				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=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(1,2,"<img src='" + data.data + "' height='200' width='200' />",app.CHAT,null);
			                     }   
			                 }
			             });
			
			        })
			
			    })
		</script>
	</body>
</html>

意思即为使用ajax访问咱们的上传文件Controller，获取上传成功后的url，将该url拼接到<img />的标签中，当称普通聊天信息发送出去便可。

用户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">
			window.CHAT = {
				socket: null,
				init: function() {
					if (window.WebSocket) {
						CHAT.socket = new WebSocket("ws://127.0.0.1:9999/ws");
						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) {
							console.log("接收到消息" + e.data);
							var receiveMsg = document.getElementById("receiveMsg");
							var html = receiveMsg.innerHTML;
							var chatMsg = JSON.parse(e.data);
							receiveMsg.innerHTML = html + "<br/>" + chatMsg.msg;
							//消息签收
							CHAT.chat(chatMsg.receiverId,null,null,app.SIGNED,String(chatMsg.msgId));
						}
					}else {
						alert("浏览器不支持WebSocket协议...");
					}
				},
				chat: function(senderId,receiverId,msg,action,extand) {
					var chatMsg = new app.ChatMsg(senderId,receiverId,msg,null);
					var dataContent = new app.DataContent(action,chatMsg,extand);
					CHAT.socket.send(JSON.stringify(dataContent));
				},
				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>
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程