怀孕日记——工作不工作

小说:怀孕日记——工作不工作作者:乙纯龙更新时间:2019-03-26字数:45787

李庆安本来是坐在裴宽主位旁边,但他坚决把位子让给了一名从河东赶来的裴家资深长辈,他的位子便转到了客人席中,和几名相国坐在一起,门下侍郎张镐是独自而来,便正好和李庆安坐在一席。

怎么确定家中的风水财位?

“异火吗?我可没有那么好的出身能有异火还能从小就得到一朵排名第四的异火。”若琳有点自嘲的说道,不过旋即双目升起了两种不同的颜色,左边深蓝色,给人一种汪洋大海般连绵不绝的感觉,右边漆黑如墨给人一种深不可测,无法琢磨的感觉。
唐虎那边闷哼一声,而正在凝聚魂力的唐三则是上身略微晃动了一下,脸色有些苍白。

这样一个惊人的数字,换做之前的丁宁,想都不敢想,可现在,它变成了现实。

引言:

在前面两篇文章中,我们对原生websocket进行了了解,且用demo来简单的讲解了其用法。但是在实际项目中,那样的用法是不可取的,理由是tomcat对高并发的支持不怎么好,特别是tomcat9之前,可以测试发现websocket连接达到的数量很低,且容易断开。
所以有现在的第三篇,对websocket的一种进阶方法。

什么是Netty

Netty是业界最流行的NIO框架之一,它的健壮性、功能、性能、可定制性和可扩展性在同类框架中都是首屈一指的,它已经得到成百上千的商用项目验证,例如Hadoop的RPC框架Avro就使用了Netty作为底层通信框架,其他还有业界主流的RPC框架,也使用Netty来构建高性能的异步通信能力。
通过对Netty的分析,我们将它的优点总结如下:

  • API使用简单,开发门槛低;
  • 功能强大,预置了多种编解码功能,支持多种主流协议;
  • 定制能力强,可以通过ChannelHandler对通信框架进行灵活地扩展;
  • 性能高,通过与其他业界主流的NIO框架对比,Netty的综合性能最优;
  • 成熟、稳定,Netty修复了已经发现的所有JDK NIO BUG,业务开发人员不需要再为NIO的BUG而烦恼;
  • 社区活跃,版本迭代周期短,发现的BUG可以被及时修复,同时,更多的新功能会加入;
  • 经历了大规模的商业应用考验,质量得到验证。Netty在互联网、大数据、网络游戏、企业应用、电信软件等众多行业已经得到了成功商用,证明它已经完全能够满足不同行业的商业应用了。

基于Netty的websocket压力测试

点此进入

Demo详解

1.导入netty包

<!-- netty -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>5.0.0.Alpha1</version>
</dependency>

 

2.server启动类
以下@Service,@PostConstruct注解是标注spring启动时启动的注解,新开一个线程去开启netty服务器端口。

package com.nettywebsocket;
import javax.annotation.PostConstruct;
import org.springframework.stereotype.Service;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
* ClassName:NettyServer 注解式随spring启动
* Function: TODO ADD FUNCTION.
* @author hxy
*/
@Service
public class NettyServer {
public static void main(String[] args) {
new NettyServer().run();
}
@PostConstruct
public void initNetty(){
new Thread(){
public void run() {
new NettyServer().run();
}
}.start();
}
public void run(){
System.out.println("===========================Netty端口启动========");
// Boss线程:由这个线程池提供的线程是boss种类的,用于创建、连接、绑定socket, (有点像门卫)然后把这些socket传给worker线程池。
// 在服务器端每个监听的socket都有一个boss线程来处理。在客户端,只有一个boss线程来处理所有的socket。
EventLoopGroup bossGroup = new NioEventLoopGroup();
// Worker线程:Worker线程执行所有的异步I/O,即处理操作
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
// ServerBootstrap 启动NIO服务的辅助启动类,负责初始话netty服务器,并且开始监听端口的socket请求
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workGroup);
// 设置非阻塞,用它来建立新accept的连接,用于构造serversocketchannel的工厂类
b.channel(NioServerSocketChannel.class);
// ChildChannelHandler 对出入的数据进行的业务操作,其继承ChannelInitializer
b.childHandler(new ChildChannelHandler());
System.out.println("服务端开启等待客户端连接 ... ...");
Channel ch = b.bind(7397).sync().channel();
ch.closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally{
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}

 

3.channle注册类

package com.nettywebsocket;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
/**
* ClassName:ChildChannelHandler
* Function: TODO ADD FUNCTION.
* @author hxy
*/
public class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel e) throws Exception {
// 设置30秒没有读到数据,则触发一个READER_IDLE事件。
// pipeline.addLast(new IdleStateHandler(30, 0, 0));
// HttpServerCodec:将请求和应答消息解码为HTTP消息
e.pipeline().addLast("http-codec",new HttpServerCodec());
// HttpObjectAggregator:将HTTP消息的多个部分合成一条完整的HTTP消息
e.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));
// ChunkedWriteHandler:向客户端发送HTML5文件
e.pipeline().addLast("http-chunked",new ChunkedWriteHandler());
// 在管道中添加我们自己的接收数据实现方法
e.pipeline().addLast("handler",new MyWebSocketServerHandler());
}
}
4.存储类
以下类是用来存储访问的channle,channelGroup的原型是set集合,保证channle的唯一,如需根据参数标注存储,可以使用currentHashMap来存储。

package com.nettywebsocket;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
/**
* ClassName:Global
* Function: TODO ADD FUNCTION.
* @author hxy
*/
public class Global {
public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}

 

5.实际处理类
以下处理类虽然做了注释,但是在这里还是详细讲解下。

  1. 这个类是单例的,每个线程处理会新实例化一个类。
  2. 每个成功的线程访问顺序:channelActive(开启连接)-handleHttpRequest(http握手处理)-messageReceived(消息接收处理)-handlerWebSocketFrame(实际处理,可以放到其他类里面分业务进行)
  3. 注意:这个demo中我做了路由功能,在handleHttpRequest中对每个channel连接的时候对每个连接的url进行绑定参数,然后在messageReceived中获取绑定的参数进行分发处理(handlerWebSocketFrame或handlerWebSocketFrame2),同时也获取了uri后置参数,有注释。
  4. 针对第三点路由分发,还有一种方法就是handshaker的uri()方法,看源码即可,简单好用。
  5. 群发的时候遍历集合或者map的时候,必须每个channle都实例化一个TextWebSocketFrame对象,否则会报错或者发不出。
package com.nettywebsocket;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
/**
* ClassName:MyWebSocketServerHandler Function: TODO ADD FUNCTION.
*
* @author hxy
*/
public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
private static final Logger logger = Logger.getLogger(WebSocketServerHandshaker.class.getName());
private WebSocketServerHandshaker handshaker;
/**
* channel 通道 action 活跃的 当客户端主动链接服务端的链接后,这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 添加
Global.group.add(ctx.channel());
System.out.println("客户端与服务端连接开启:" + ctx.channel().remoteAddress().toString());
}
/**
* channel 通道 Inactive 不活跃的 当客户端主动断开服务端的链接后,这个通道就是不活跃的。也就是说客户端与服务端关闭了通信通道并且不可以传输数据
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
// 移除
Global.group.remove(ctx.channel());
System.out.println("客户端与服务端连接关闭:" + ctx.channel().remoteAddress().toString());
}
/**
* 接收客户端发送的消息 channel 通道 Read 读 简而言之就是从通道中读取数据,也就是服务端接收客户端发来的数据。但是这个数据在不进行解码时它是ByteBuf类型的
*/
@Override
protected void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
// 传统的HTTP接入
if (msg instanceof FullHttpRequest) {
handleHttpRequest(ctx, ((FullHttpRequest) msg));
// WebSocket接入
} else if (msg instanceof WebSocketFrame) {
System.out.println(handshaker.uri());
if("anzhuo".equals(ctx.attr(AttributeKey.valueOf("type")).get())){
handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
}else{
handlerWebSocketFrame2(ctx, (WebSocketFrame) msg);
}
}
}
/**
* channel 通道 Read 读取 Complete 完成 在通道读取完成后会在这个方法里通知,对应可以做刷新操作 ctx.flush()
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
// 判断是否关闭链路的指令
if (frame instanceof CloseWebSocketFrame) {
System.out.println(1);
handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
return;
}
// 判断是否ping消息
if (frame instanceof PingWebSocketFrame) {
ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
return;
}
// 本例程仅支持文本消息,不支持二进制消息
if (!(frame instanceof TextWebSocketFrame)) {
System.out.println("本例程仅支持文本消息,不支持二进制消息");
throw new UnsupportedOperationException(
String.format("%s frame types not supported", frame.getClass().getName()));
}
// 返回应答消息
String request = ((TextWebSocketFrame) frame).text();
System.out.println("服务端收到:" + request);
if (logger.isLoggable(Level.FINE)) {
logger.fine(String.format("%s received %s", ctx.channel(), request));
}
TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + ":" + request);
// 群发
Global.group.writeAndFlush(tws);
// 返回【谁发的发给谁】
// ctx.channel().writeAndFlush(tws);
}
private void handlerWebSocketFrame2(ChannelHandlerContext ctx, WebSocketFrame frame) {
// 判断是否关闭链路的指令
if (frame instanceof CloseWebSocketFrame) {
handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
return;
}
// 判断是否ping消息
if (frame instanceof PingWebSocketFrame) {
ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
return;
}
// 本例程仅支持文本消息,不支持二进制消息
if (!(frame instanceof TextWebSocketFrame)) {
System.out.println("本例程仅支持文本消息,不支持二进制消息");
throw new UnsupportedOperationException(
String.format("%s frame types not supported", frame.getClass().getName()));
}
// 返回应答消息
String request = ((TextWebSocketFrame) frame).text();
System.out.println("服务端2收到:" + request);
if (logger.isLoggable(Level.FINE)) {
logger.fine(String.format("%s received %s", ctx.channel(), request));
}
TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + ":" + request);
// 群发
Global.group.writeAndFlush(tws);
// 返回【谁发的发给谁】
// ctx.channel().writeAndFlush(tws);
}
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
// 如果HTTP解码失败,返回HHTP异常
if (!req.getDecoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
sendHttpResponse(ctx, req,
new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
return;
}
//获取url后置参数
HttpMethod method=req.getMethod();
String uri=req.getUri();
QueryStringDecoder queryStringDecoder = new QueryStringDecoder(uri);
Map<String, List<String>> parameters = queryStringDecoder.parameters();
System.out.println(parameters.get("request").get(0));
if(method==HttpMethod.GET&&"/webssss".equals(uri)){
//....处理
ctx.attr(AttributeKey.valueOf("type")).set("anzhuo");
}else if(method==HttpMethod.GET&&"/websocket".equals(uri)){
//...处理
ctx.attr(AttributeKey.valueOf("type")).set("live");
}
// 构造握手响应返回,本机测试
WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
"ws://"+req.headers().get(HttpHeaders.Names.HOST)+uri, null, false);
handshaker = wsFactory.newHandshaker(req);
if (handshaker == null) {
WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
} else {
handshaker.handshake(ctx.channel(), req);
}
}
private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
// 返回应答给客户端
if (res.getStatus().code() != 200) {
ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
res.content().writeBytes(buf);
buf.release();
}
// 如果是非Keep-Alive,关闭连接
ChannelFuture f = ctx.channel().writeAndFlush(res);
if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
f.addListener(ChannelFutureListener.CLOSE);
}
}
/**
* exception 异常 Caught 抓住 抓住异常,当发生异常的时候,可以做一些相应的处理,比如打印日志、关闭链接
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

 

  • 以上就是netty-websocket的Demo了,应该已经解释的很详细了,同时应对的并发量也满足一般企业用于websocket的连接,如果需要不够,可以用nginx负载均衡增加。
  • 最后给大家一条建议,在实际项目中,别让这种长连接一直保持,在nginx中可以设置连接无交流超时断开,大概设置10分钟左右,然后每8分钟定时从服务端发送一条心跳,具体想法就看你们喽~


  亲们,请点此链接,博主自己的博客更需要你们的光临。  时光盒

编辑:卓道文乙

发布:2019-03-26 02:20:36

当前文章:http://adsl66.com/t3ye2/59513.html

38D、梦姑、倒追女与万能USB 原创:面对灾难最好的帮助—倾听篇 跑步=生小孩,是男人就该为女人跑一次步! 失恋后不相信爱情? “开门,我是警察”,这时不是开门,而是要做22件事 为什么唾液不能止渴? 罗李华谈:属狗的人2016年运程 经典原创禅诗:【禅.十字回文诗】

16285 15738 38092 72722 75375 84662 49256 72706 83728 87031 33804 63863 59992 88506 55590 80362 39083 29561 50452 29331

我要说两句: (0人参与)

发布