功能介绍Netty开发服务器
HTML实现客户端
实现服务端与客户端时实时交互
开发步骤
1.导包
io.netty
netty‐all
5.0.0.Alpha2
2.工程配置文件：NettyConfig
/**
* 这里放的是工程中相应的配置
*/
public class NettyConfig{
/**
* 用于存储每一个客户端接入进来时的channel对象
*/
public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}
3.MyWebSocketHandler
public class MyWebSocketHandler extends SimpleChannelInboundHandler { private WebSocketServerHandshaker handshaker;
//请求路径
private static final String WEB_SOCKET_URL = "ws://localhost:8888/websocket";
//工程出现异常的时候调用@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace(); ctx.close(); //关闭当前连接
}
//客户端与服务器创建连接的时候调用@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception { NettyConfig.group.add(ctx.channel());
System.out.println("客户端与服务器连接开启");
}
//客户端与服务器断开连接的时候调用@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception { NettyConfig.group.remove(ctx.channel());
System.out.println("客户端与服务器断开连接");
}
//服务端接收客户端发送过来的数据结束之后调用@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception{ ctx.flush(); //进行数据清除
}
//服务端处理客户端调用的核心方法
protected void messageReceived(ChannelHandlerContext context, Object msg) throws Exception {
//处理客户端向服务端发起http握手请求业务if(msg instanceof FullHttpRequest){
handHttpRequest(context,(FullHttpRequest) msg);
}else if(msg instanceof WebSocketFrame){
//处理websocket连接业务handWebSocketFrme(context,(WebSocketFrame)msg);
}
}
/**
* 处理客户端与服务端之前的websocket业务
* @param ctx
* @param frame
*/
private void handWebSocketFrme(ChannelHandlerContext ctx,WebSocketFrame frame){
//判断是否是关闭websocket的指令
if(frame instanceof CloseWebSocketFrame){ handshaker.close(ctx.channel(),((CloseWebSocketFrame) frame).retain());
}
//判断是否是Ping消息
if(frame instanceof PingWebSocketFrame){
//返回一个胖信息
ctx.channel().write(newPongWebSocketFrame(frame.content().retain())); return;
}
//判断是否是二进制消息，如果是二进制消息，抛出异常if(!(frame instanceof TextWebSocketFrame)){
System.out.println("目前我们不支持二进制消息");
throw new RuntimeException(this.getClass().getName()+"不支持的消息");
}
//返回应答消息
//获取客户端向服务端发送的消息
String request = ((TextWebSocketFrame)frame).text(); System.out.println("服务端收到客户端的消息===>"+request); TextWebSocketFrame tws = new TextWebSocketFrame(new
Date().toString()+ctx.channel().id()+"===>"+request);
//群发，服务端向每个上来的客户端群发消息NettyConfig.group.writeAndFlush(tws);
}
/**
* 处理客户端向服务器发起http握手请求的业务
* @param ctx
* @param req
*/
private void handHttpRequest(ChannelHandlerContext ctx,FullHttpRequest req){
// req.headers().get()
// req.decoderResult().isSuccess()
//不是websocket,就不是客户端发给服务器的Http请求if(!req.decoderResult().isSuccess()|| !
("websocket").equals(req.headers().get("Upgrade"))){
sendHttpResponse(ctx,req,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
return;
}
WebSocketServerHandshakerFactory wsFactory =
newWebSocketServerHandshakerFactory(WEB_SOCKET_URL,null,false); handshaker = wsFactory.newHandshaker(req);
if(handshaker==null){
WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
}else{
handshaker.handshake(ctx.channel(),req);
}
}
/**
* 服务端向客户端响应消息
* @param ctx
* @param req
* @param res
*/
private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res){
if(res.status().code()!=200){
ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
res.content().writeBytes(buf); buf.release();
}
//服务器向客户端发送数据
ChannelFuture f = ctx.channel().writeAndFlush(res); if(res.status().code()!=200){
f.addListener(ChannelFutureListener.CLOSE);
}
}
}
4 初始化连接各个组件:MyWebSocketChannelHandler
/**
* 初始化连接各个组件
*/
public class MyWebSocketChannelHandlerextends ChannelInitializer{
protected void initChannel(SocketChannel e) throws Exception { e.pipeline().addLast("http‐codec",new HttpServerCodec()); e.pipeline().addLast("aggregator",new HttpObjectAggregator(66636)); e.pipeline().addLast("http‐chunked",new ChunkedWriteHandler()); e.pipeline().addLast("handler",new MyWebSocketHandler());
}
}
5 程序入口
/**
* 程序的入口，负责启动应用
*/
public class Main {
public static void main(String[] args) { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workGroup = new NioEventLoopGroup();try{
ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup,workGroup); b.channel(NioServerSocketChannel.class); b.childHandler(new MyWebSocketChannelHandler());
System.out.println("服务端开启等待客户端连接");
Channel ch = b.bind(8888).sync().channel(); ch.closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
//优雅的退出程序bossGroup.shutdownGracefully(); workGroup.shutdownGracefully();
}
}
}
6 web端测试代码
WebSocket客户端var socket; if(!window.WebSocket){
window.WebSocket = window.MozWebSocket;
}
if(window.WebSocket){
socket = new WebSocket("ws://localhost:8888/websocket"); socket.onmessage = function (event) {
var ta = document.getElementById("responseContext"); ta.value += event.data +"\r\n";
}
socket.onopen = function (event) {
var ta = document.getElementById("responseContext"); ta.value = "您浏览器支持webSoceket，请进行后续 操作\r\n";
}
socket.onclose = function (event) {
var ta = document.getElementById("responseContext"); ta.value = "";
ta.value = "webSoceket连接已经关闭";
}
}else{
console.debug("浏览器不支持webSocket")
}
function send() {
var message = document.getElementById("message").value; if(!window.WebSocket){
return;
}
if(socket.readyState == WebSocket.OPEN){ socket.send(message)
}else{
alert("连接没有建立成功！");
}
}
