Netty实现Http服务
主要的变化是在初始化器中引入了新的编解码器
一些创建的类作用和Netty HelloWorld的小demo一样我这里就不再次重复了
1、Http服务端代码
public class HttpServer {
public static void main(String[] args) {
// 创建Reactor
// 用来管理channel 监听事件 ,是无限循环的事件组(线程池)
EventLoopGroup bossLoopGroup = new NioEventLoopGroup();
EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
// 服务端的启动对象
ServerBootstrap serverBootstrap = new ServerBootstrap();
// 设置相关参数 这是一个链式编程
serverBootstrap.group(bossLoopGroup,workerLoopGroup)
// 声明通道类型
.channel(NioServerSocketChannel.class)
// 设置处理器 我这里设置了netty提供的Handler 处理器
.handler(new LoggingHandler(LogLevel.INFO))
// 定义客户连接端处理器的使用
// ChannelInitializer 通道处理化
// 可以自定义通道初始化器,如实现编码解码器时
.childHandler(new ChannelInitializer<SocketChannel>() {
protected void initChannel(SocketChannel ch) throws Exception {
// 需要处理的是客户端通道
// 通道代表的是 连接的角色 管道代表的是 处理业务的逻辑管理
// 管道相当与一个链表, 将不同的处理器连接起来,管理的是处理器的顺序
ch.pipeline().addLast(new HttpMyInitializer());
}
});
System.out.println("服务端初始化完成");
// 启动需要设置端口 还需要设置是异步启动
try {
// 设置异步的future
ChannelFuture future = serverBootstrap.bind(9988).sync();
// 将关闭的通道也设置成异步的
// 阻塞finally 中的代码
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
// 优雅关闭
bossLoopGroup.shutdownGracefully();
workerLoopGroup.shutdownGracefully();
}
}
}
1.1 Http服务自定义初始化器
下面是需要了解的组件
请求和响应的编码解码器:
客户端
HttpResponseDecoder 解码器,
处理服务端的响应(客户端)
HttpRequestEncoder 编码器,
处理服务端的请求(客户端)
服务端
HttpRequestDecoder 解码器,
处理客户端的请求(服务端)
HttpResponseEncoder 编码器,
处理客户端的响应(服务端)
由于上面的编码解码器都比较绕,所以还有两个组合的类提供
HttpClientCodeC :
编码解码器,用于客户端 HttpResponseDecoder + HttpRequestEncoder
HttpServerCodeC:
编码解码器,用于服务端 HttpRequestDecoder + HttpResponseEncoder
聚合
由于http的请求和响应,可能由很多部分组成,需要聚合成一个完整的消息
HttpObjectAggregator -> FullHttpRequest / FullHttpResponse
压缩
由于网络上有些情况文件或者图片需要压缩,所以需要压缩处理器
HttpContentCompressor 压缩,用于服务端
HttpContentDecompressor 解压缩,用于客户端
自定义初始化器HttpMyInitializer 需要继承ChannelInitializer泛型是Channel
public class HttpMyInitializer extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 先解码后编码
// pipeline.addLast("decoder",new HttpRequestDecoder());
// pipeline.addLast("encoder",new HttpResponseEncoder());
// 相当于上面两行
pipeline.addLast("codec",new HttpServerCodec());
// 压缩数据
pipeline.addLast("compressor",new HttpContentCompressor());
// 聚合成完整的消息 参数代表处理的最大值
pipeline.addLast("aggregator",new HttpObjectAggregator(512 * 1024));
// 添加处理器
pipeline.addLast(new MyHttpHandler());
}
}
1.2 Http服务自定义处理器
需要继承SimpleChannelInboundHandler类注意的是泛型需要定义为 FullHttpRequest
/**
* 泛型需要定义为 FullHttpRequest
*
*/
public class MyHttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
/**
*
* @param ctx 通道处理器上下文
* @param msg 接收客户端数据消息
* @throws Exception
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
// 设定 版本 、响应码、响应的数据(ByteBuf) 等
DefaultFullHttpResponse response =new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer("http hello netty demo ".getBytes()));
// 还需要设置响应头 HttpHeaders 来接收
// 设置字段名 使用HttpHeaderNames ,字段值使用HttpHeaderValues
HttpHeaders headers = response.headers();
//content/type ;text/plain
headers.add(HttpHeaderNames.CONTENT_TYPE,HttpHeaderValues.TEXT_PLAIN+"charset=UTF-8");
// 设置包的大小时, 调用 readableBytes方法
headers.add(HttpHeaderNames.CONTENT_LENGTH,response.content().readableBytes());
// 将response 写入通道 这里不用writeAndFlush方法, 而是在channelReadComplete读完成的方法内来刷新通道
ctx.write(response);
}
/**
* 用来刷新channelRead0 写入通道里面的response 数据
* @param ctx
* @throws Exception
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
1.3 Http服务最后展示结果
启动服务端、客户端我展示代码,可以随便启动一个我之前的小demo客户端记得改端口9988就行
访问localhost:9988
在Edge浏览器展示
Netty实现WebSocket服务
http协议的缺陷: 通信只能由客户端发起。需要一种服务端能够主动推送的能力—websocket。这种双向通信的能力,也叫“全双工”。
协议标识符: http://127.0.0.1/ -> ws://127.0.0.1/
通信的最小单位是帧frame。
2、WebSocket服务服务端代码
同样的配方,大同小异, 只是初始化器和处理器不同,需要自定义
public class WebSocketServer {
public static void main(String[] args) {
// 创建Reactor
// 用来管理channel 监听事件 ,是无限循环的事件组(线程池)
EventLoopGroup bossLoopGroup = new NioEventLoopGroup();
EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
// 服务端的启动对象
ServerBootstrap serverBootstrap = new ServerBootstrap();
// 设置相关参数 这是一个链式编程
serverBootstrap.group(bossLoopGroup,workerLoopGroup)
// 声明通道类型
.channel(NioServerSocketChannel.class)
// 设置处理器 我这里设置了netty提供的Handler 处理器
.handler(new LoggingHandler(LogLevel.INFO))
// 定义客户连接端处理器的使用
// ChannelInitializer 通道处理化
// 可以自定义通道初始化器,如实现编码解码器时
.childHandler(new WebSocketInitializer());
System.out.println("服务端初始化完成");
// 启动需要设置端口 还需要设置是异步启动
try {
// 设置异步的future
ChannelFuture future = serverBootstrap.bind(7777).sync();
// 将关闭的通道也设置成异步的
// 阻塞finally 中的代码
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
// 优雅关闭
bossLoopGroup.shutdownGracefully();
workerLoopGroup.shutdownGracefully();
}
}
}
2.1 WebSocket服务自定义初始化器
继承ChannelInitializer 泛型是SocketChannel
public class WebSocketInitializer extends ChannelInitializer<SocketChannel> {
/**
*
* @param ch
* @throws Exception
*/
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 设置编码解码器
pipeline.addLast(new HttpServerCodec());
// 用于处理较大的数据
pipeline.addLast(new ChunkedWriteHandler());
// 设置聚合器
pipeline.addLast(new HttpObjectAggregator(512 * 1024));
// 声明请求路径 ws://127.0.0.1:7777/hello
pipeline.addLast(new WebSocketServerProtocolHandler("/hello"));
// 自定义处理器
pipeline.addLast(new WebSocketHandler());
}
}
2.2 WebSocket服务自定义处理器
主要的是channelRead0方法
/**
* 本次业务处理的数据是文本, WebSocket通信是通过帧来传输
* 所以泛型为 TextWebSocketFrame
*/
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
// 当多个通道传入handler , 使用通道组的管理方法
// GlobalEventExecutor 全局事件执行器
//INSTANCE 代表的是单例
private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
/**
*
* @param ctx 通道处理器上下文
* @param msg 文本消息帧
* @throws Exception
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
// 浏览器返回的信息帧
System.out.println("msg:"+msg.text());
Channel channel = ctx.channel();
// 需要响应会浏览器的信息, 需要是TextWebSocketFrame 类型
TextWebSocketFrame webSocketFrame = new TextWebSocketFrame(ctx.channel().remoteAddress()+"客户端:"+msg.text()+"\r\n");
channel.writeAndFlush(webSocketFrame);
}
/**
* 连接成功, 此时通道是活跃的时候触发
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
LocalDate today = LocalDate.now();
String dateStr = today.toString(); // 默认格式为 "yyyy-MM-dd"
ctx.writeAndFlush("Welcome to server-- now :"+dateStr+"\r\n");
}
/**
* 通道不活跃 ,用于处理用户下线的逻辑
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println(ctx.channel().remoteAddress()+"下线了\r\n");
}
/**
*
* @param ctx 通道处理器上下文
* @throws Exception
* 连接刚刚建立时 ,第一个被执行的方法,
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
System.out.println("[服务端地址]:"+ctx.channel().remoteAddress()+"连接成功\r\n");
// 添加到通道组中管理
channelGroup.add(ctx.channel());
}
/**
*
* @param ctx 通道处理器上下文
* @throws Exception
* 当连接断开 最后执行的方法
* 连接断开时 , channel 会自动从 通道组中移除
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
System.out.println("[服务端地址]:"+ctx.channel().remoteAddress()+"断开连接\r\n");
}
/**
* 通用异常处理类
* @param ctx 通道处理器上下文
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// 关闭
ctx.close();
}
}
2.3 WebSocket服务前端界面
实现一个聊天的小demo
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Hello WebSocket</title>
</head>
<body>
<script>
var socket;
// 判断当前浏览器是否支持websockt
if (!window.WebSocket) {
alert("不支持websocket")
} else {
socket = new WebSocket("ws://127.0.0.1:7777/hello");
// 设置开启连接的方法
socket.onopen = function (ev) {
var tmp = document.getElementById("respText");
tmp.value = "连接已开启";
}
// 设置关闭连接的方法
socket.onclose = function (ev) {
var tmp = document.getElementById("respText");
tmp.value = tmp.value + "\n" + "连接已关闭";
}
// 设置接收数据的方法
socket.onmessage = function (ev) {
var tmp = document.getElementById("respText");
tmp.value = tmp.value + "\n" + ev.data;
}
}
function send(message) {
// 先判断socket是否已经创建
if (!window.socket) {
return
}
// 判断socket的状态
// CONNECTING 正在连接 CLOSING 正在关闭
// CLOSED 已经关闭或打开连接失败
// OPEN 连接成功 可以正常通信
if (socket.readyState == WebSocket.OPEN) {
socket.send(message);
} else {
alert("连接未开启");
}
}
</script>
<!--防止表单自动提交-->
<form onsubmit="return false">
<textarea name="message" style="height: 400px;width: 400px"></textarea>
<input type="button" value="发送" onclick="send(this.form.message.value)">
<textarea id="respText" style="height: 400px;width: 400px"></textarea>
</form>
</body>
</html>
2.3 WebSocket结果展示
启动WebSocke服务器,运行前端代码如下如所示: