From 805a1aa3fb86c7829e43ae2a70466e7e52293ca6 Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Sun, 8 Aug 2021 18:00:27 +0800 Subject: [PATCH] =?UTF-8?q?perf[net]:=20=E6=94=AF=E6=8C=81http=E5=8D=8F?= =?UTF-8?q?=E8=AE=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/zfoo/net/core/AbstractServer.java | 2 +- .../com/zfoo/net/core/http/HttpServer.java | 61 ++++++++++ .../java/com/zfoo/net/core/udp/UdpServer.java | 6 + .../net/core/websocket/WebsocketServer.java | 14 +-- .../dispatcher/manager/PacketDispatcher.java | 3 - .../handler/codec/http/HttpCodecHandler.java | 115 ++++++++++++++++++ .../packet/model/HttpPacketAttachment.java | 47 +++++++ .../net/packet/model/NoAnswerAttachment.java | 2 +- .../packet/model/PacketAttachmentType.java | 21 ++-- net/src/test/resources/protocol.xml | 3 +- 10 files changed, 250 insertions(+), 24 deletions(-) create mode 100644 net/src/main/java/com/zfoo/net/core/http/HttpServer.java create mode 100644 net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java create mode 100644 net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java diff --git a/net/src/main/java/com/zfoo/net/core/AbstractServer.java b/net/src/main/java/com/zfoo/net/core/AbstractServer.java index dd004724..35f894f7 100644 --- a/net/src/main/java/com/zfoo/net/core/AbstractServer.java +++ b/net/src/main/java/com/zfoo/net/core/AbstractServer.java @@ -94,7 +94,7 @@ public abstract class AbstractServer implements IServer { allServers.add(this); - logger.info("TcpServer started at [{}:{}]", hostAddress, port); + logger.info("{} started at [{}:{}]", this.getClass().getSimpleName(), hostAddress, port); } diff --git a/net/src/main/java/com/zfoo/net/core/http/HttpServer.java b/net/src/main/java/com/zfoo/net/core/http/HttpServer.java new file mode 100644 index 00000000..36eeb442 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/http/HttpServer.java @@ -0,0 +1,61 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + * + */ + +package com.zfoo.net.core.http; + +import com.zfoo.net.core.AbstractServer; +import com.zfoo.net.handler.ServerDispatcherHandler; +import com.zfoo.net.handler.codec.http.HttpCodecHandler; +import com.zfoo.protocol.IPacket; +import com.zfoo.util.net.HostAndPort; +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.stream.ChunkedWriteHandler; + +import java.util.function.Function; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class HttpServer extends AbstractServer { + + /** + * http的地址解析器 + */ + private Function uriResolver; + + public HttpServer(HostAndPort host, Function uriResolver) { + super(host); + this.uriResolver = uriResolver; + } + + @Override + public ChannelInitializer channelChannelInitializer() { + return new ChannelHandlerInitializer(); + } + + + private class ChannelHandlerInitializer extends ChannelInitializer { + @Override + protected void initChannel(SocketChannel channel) { + channel.pipeline().addLast(new HttpServerCodec()); + channel.pipeline().addLast(new ChunkedWriteHandler()); + channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024)); + channel.pipeline().addLast(new HttpCodecHandler(uriResolver)); + channel.pipeline().addLast(new ServerDispatcherHandler()); + } + } +} diff --git a/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java b/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java index 4b0a2506..52c105b7 100644 --- a/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java +++ b/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java @@ -26,6 +26,8 @@ import io.netty.channel.epoll.EpollEventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioDatagramChannel; import io.netty.util.concurrent.DefaultThreadFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @author jaysunxiao @@ -33,6 +35,8 @@ import io.netty.util.concurrent.DefaultThreadFactory; */ public class UdpServer extends AbstractServer { + private static final Logger logger = LoggerFactory.getLogger(UdpServer.class); + public UdpServer(HostAndPort host) { super(host); } @@ -58,6 +62,8 @@ public class UdpServer extends AbstractServer { channel = channelFuture.channel(); allServers.add(this); + + logger.info("{} started at [{}:{}]", this.getClass().getSimpleName(), hostAddress, port); } @Override diff --git a/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java b/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java index ac59c94d..ac9684af 100644 --- a/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java +++ b/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java @@ -18,7 +18,6 @@ import com.zfoo.net.handler.ServerDispatcherHandler; import com.zfoo.net.handler.codec.websocket.WebSocketCodecHandler; import com.zfoo.util.net.HostAndPort; import io.netty.channel.ChannelInitializer; -import io.netty.channel.ChannelPipeline; import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; @@ -45,19 +44,18 @@ public class WebsocketServer extends AbstractServer { @Override public void initChannel(SocketChannel channel) { - ChannelPipeline pipeline = channel.pipeline(); // 编解码 http 请求 - pipeline.addLast(new HttpServerCodec()); + channel.pipeline().addLast(new HttpServerCodec()); // 写文件内容，支持异步发送大的码流，一般用于发送文件流 - pipeline.addLast(new ChunkedWriteHandler()); + channel.pipeline().addLast(new ChunkedWriteHandler()); // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest // 保证接收的 Http 请求的完整性 - pipeline.addLast(new HttpObjectAggregator(64 * 1024)); + channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024)); // 处理其他的 WebSocketFrame - pipeline.addLast(new WebSocketServerProtocolHandler("/websocket")); + channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket")); // 编解码WebSocketFrame二进制协议 - pipeline.addLast(new WebSocketCodecHandler()); - pipeline.addLast(new ServerDispatcherHandler()); + channel.pipeline().addLast(new WebSocketCodecHandler()); + channel.pipeline().addLast(new ServerDispatcherHandler()); } } diff --git a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java index 4687ec84..0a842108 100644 --- a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java +++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java @@ -129,9 +129,6 @@ public class PacketDispatcher implements IPacketDispatcher { } return; } - break; - case NORMAL_PACKET: - break; default: break; diff --git a/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java b/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java new file mode 100644 index 00000000..7d06ca1f --- /dev/null +++ b/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java @@ -0,0 +1,115 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.handler.codec.http; + +import com.zfoo.net.NetContext; +import com.zfoo.net.packet.model.DecodedPacketInfo; +import com.zfoo.net.packet.model.EncodedPacketInfo; +import com.zfoo.net.packet.model.HttpPacketAttachment; +import com.zfoo.net.packet.service.PacketService; +import com.zfoo.net.util.SessionUtils; +import com.zfoo.protocol.IPacket; +import com.zfoo.protocol.util.JsonUtils; +import com.zfoo.protocol.util.StringUtils; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.MessageToMessageCodec; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpVersion; +import io.netty.util.ReferenceCountUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.function.Function; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class HttpCodecHandler extends MessageToMessageCodec { + + private static final Logger logger = LoggerFactory.getLogger(HttpCodecHandler.class); + + private Function uriResolver; + + public HttpCodecHandler(Function uriResolver) { + super(); + this.uriResolver = uriResolver; + } + + @Override + protected void decode(ChannelHandlerContext channelHandlerContext, FullHttpRequest fullHttpRequest, List list) { + var uri = fullHttpRequest.uri(); + + ByteBuf in = fullHttpRequest.content(); + + // 不够读一个int + if (in.readableBytes() <= PacketService.PACKET_HEAD_LENGTH) { + return; + } + + in.markReaderIndex(); + var length = in.readInt(); + + // 如果长度非法，则抛出异常断开连接 + if (length < 0) { + throw new IllegalArgumentException(StringUtils.format("[session:{}]的包头长度[length:{}]非法" + , SessionUtils.sessionInfo(channelHandlerContext), length)); + } + + // ByteBuf里的数据太小 + if (in.readableBytes() < length) { + in.resetReaderIndex(); + return; + } + + ByteBuf tmpByteBuf = null; + try { + tmpByteBuf = in.readRetainedSlice(length); + DecodedPacketInfo packetInfo = NetContext.getPacketService().read(tmpByteBuf); + + packetInfo.setPacketAttachment(HttpPacketAttachment.valueOf()); + list.add(packetInfo); + } catch (Exception e) { + logger.error("exception异常", e); + throw e; + } catch (Throwable t) { + logger.error("throwable错误", t); + throw t; + } finally { + ReferenceCountUtil.release(tmpByteBuf); + } + } + + @Override + protected void encode(ChannelHandlerContext channelHandlerContext, EncodedPacketInfo out, List list) { + try { + var byteBuf = channelHandlerContext.alloc().ioBuffer(); + var httpPacketAttachment = (HttpPacketAttachment) out.getPacketAttachment(); + + var fullHttpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer("I am ok".getBytes())); + + list.add(fullHttpResponse); + } catch (Exception e) { + logger.error("[{}]编码exception异常", JsonUtils.object2String(out), e); + throw e; + } catch (Throwable t) { + logger.error("[{}]编码throwable错误", JsonUtils.object2String(out), t); + throw t; + } + } +} diff --git a/net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java b/net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java new file mode 100644 index 00000000..3c6feaa7 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java @@ -0,0 +1,47 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.packet.model; + +import com.zfoo.util.math.RandomUtils; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class HttpPacketAttachment implements IPacketAttachment { + + public static final transient short PROTOCOL_ID = 3; + + + public static HttpPacketAttachment valueOf() { + var attachment = new HttpPacketAttachment(); + return attachment; + } + + @Override + public PacketAttachmentType packetType() { + return PacketAttachmentType.HTTP_PACKET; + } + + @Override + public int executorConsistentHash() { + return RandomUtils.randomInt(); + } + + @Override + public short protocolId() { + return PROTOCOL_ID; + } + +} diff --git a/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java b/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java index de118d08..ef2f31fd 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java +++ b/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java @@ -21,7 +21,7 @@ package com.zfoo.net.packet.model; */ public class NoAnswerAttachment implements IPacketAttachment { - public static final transient short PROTOCOL_ID = 3; + public static final transient short PROTOCOL_ID = 4; /** * 用来在TaskManage中计算一致性hash的参数 diff --git a/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java b/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java index 95dd66d7..193f5328 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java +++ b/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java @@ -22,32 +22,33 @@ import java.util.Map; */ public enum PacketAttachmentType { - /** - * 正常的附加包 - */ - NORMAL_PACKET((byte) 0, null), - /** * 带有同步或者异步信息的附加包 */ - SIGNAL_PACKET((byte) 1, SignalPacketAttachment.class), + SIGNAL_PACKET((byte) 0, SignalPacketAttachment.class), /** * 带有网关信息的附加包 */ - GATEWAY_PACKET((byte) 2, GatewayPacketAttachment.class), + GATEWAY_PACKET((byte) 1, GatewayPacketAttachment.class), /** * udp消息的附加包 */ - UDP_PACKET((byte) 3, UdpPacketAttachment.class), + UDP_PACKET((byte) 2, UdpPacketAttachment.class), + + + /** + * http消息的附加包 + */ + HTTP_PACKET((byte) 3, HttpPacketAttachment.class), /** * 无返回消息的附加包 */ NO_ANSWER_PACKET((byte) 4, NoAnswerAttachment.class), - + ; @@ -60,7 +61,7 @@ public enum PacketAttachmentType { } public static PacketAttachmentType getPacketType(byte packetType) { - return map.getOrDefault(packetType, PacketAttachmentType.NORMAL_PACKET); + return map.getOrDefault(packetType, PacketAttachmentType.NO_ANSWER_PACKET); } public byte getPacketType() { diff --git a/net/src/test/resources/protocol.xml b/net/src/test/resources/protocol.xml index b682aa6e..ecd3f484 100644 --- a/net/src/test/resources/protocol.xml +++ b/net/src/test/resources/protocol.xml @@ -7,7 +7,8 @@ - + +