无可致疑,netty是java的网络通讯框架,支持高并发。本文扫描使用netty完成简单的http的能力,不涉及安全,业务过滤等内容。
片段1
/** * 启动http服务器 * @throws InterruptedException */ private void runHttpServer(final EventProducer evtProducer) throws InterruptedException { // 配置TCP服务器. EventLoopGroup bossGroup = new NioEventLoopGroup(ServerBootOption.Parent_EventLoopGroup_ThreadNum); EventLoopGroup workerGroup = new NioEventLoopGroup(ServerBootOption.Child_EventLoopGroup_ThreadNum); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 10240) .option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) .option(ChannelOption.SO_REUSEADDR, true) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.SO_LINGER, 0) //.childOption(ChannelOption.SO_TIMEOUT, ServerBootOption.SO_TIMEOUT) // .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>(){ @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline p = ch.pipeline(); p.addLast("idleStateHandler", new IdleStateHandler(60, 60, 30));<span style="color:#FF0000;">//读信道空闲60s,写信道空闲60s,读,写信道空闲30s</span> p.addLast("http_server_codec", new HttpServerCodec());//<span style="color:#FF0000;">http消息转换</span> p.addLast("http_server_handler",new HttpProxyServerHandler(manager,evtProducer));//<span style="color:#FF0000;">消息处理器</span> } }); // Start the tcp server. ChannelFuture f = b.bind(new InetSocketAddress(ip, port)).sync();//<span style="color:#FF0000;">启动http服务进程</span> logger.info("start http server ok"); // Wait until the server socket is closed. f.channel().closeFuture().sync(); } finally { // Shut down all event loops to terminate all threads. logger.info("Shut down all event loops to terminate all threads."); bossGroup.shutdownGracefully();//关闭服务进程 workerGroup.shutdownGracefully();//关闭服务进程 } }片段2HttpProxyServerHandler
/** * * @author * http 请求事件处理器,负责分发http请求事件 */ public class HttpProxyServerHandler extends ChannelHandlerAdapter{ private static final Logger logger = Logger.getLogger(HttpProxyServerHandler.class); private SessionContextManager manager;<span style="color:#FF0000;">//会话管理</span> private final AttributeKey<Long> timestamp = AttributeKey.valueOf("timestamp"); private final StringBuilder buf = new StringBuilder(); public HttpProxyServerHandler(SessionContextManager manager){ this.manager=manager; } @Override public void handlerAdded(final ChannelHandlerContext ctx) throws Exception{ logger.info("["+ctx.channel().id().asLongText()+" ] is Added "); } @Override <span style="color:#FF0000;">会话关闭,失效事件</span> public void channelInactive(ChannelHandlerContext ctx) throws Exception { manager.getSession().remove(ctx.channel().id().asLongText()); super.channelInactive(ctx); } @Override <span style="color:#FF0000;">读消息</span><span style="color:#FF0000;">事件,业务处理的入口</span> public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception { logger.info("HttpProxyServerHandler[channelRead] "); Attribute<Long> attr_timestamp = ctx.channel().attr(timestamp); attr_timestamp.set(System.currentTimeMillis());//<span style="color:#FF0000;">测试信道中保存状态消息</span>,<span style="color:#FF0000;">如消息处理的开如时间</span> SessionContext sctx=new SessionContext(); sctx.setCtx(ctx); sctx.setChannelId(ctx.channel().id()); sctx.setSessionActiveTime(""+System.currentTimeMillis()); manager.getSession().put(sctx.getChannelId().asLongText(), sctx);//<span style="color:#FF0000;">manager.getSession() 为并发Map结构,用于会话保持</span> logger.info(sctx.getChannelId().asLongText()+" req time "+System.currentTimeMillis()); try { dispatchMeesage(ctx,manager ,msg); } finally { ReferenceCountUtil.release(msg); } } private QueryStringDecoder qdecoder = null; private String uri="";//http uri private String url="";//http url private Map<String,List<String>> httpReqParams=null;//http请求参数 /** * 消息分发 * @param ctx * @param manager * @param msg 消息 * @return * @throws Exception */ private String dispatchMeesage(final ChannelHandlerContext ctx, final SessionContextManager manager, final Object msg) throws Exception{ String decode_message = ""; HttpResponseUtils util = new HttpResponseUtils(); String result_code="1"; if (msg instanceof HttpRequest) {// http请求头 HttpRequest req = (HttpRequest) msg; url = req.getUri(); if (url.equals("/favicon.ico")) { ctx.close(); return "0"; } if(!url.equals("/billing")){ ctx.close(); return "0"; } //if (is100ContinueExpected(req)) { // ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE)); // } qdecoder = new QueryStringDecoder(url); httpReqParams = qdecoder.parameters(); uri = qdecoder.path(); /* TODO:身份认证 * if(qdecoder.parameters().containsKey("crendial")){ * crendial=(String * )qdecoder.parameters().get("crendial").get(0).toUpperCase(); } */ } else if (msg instanceof HttpContent) { // http请求体 HttpContent httpContent = (HttpContent) msg; ByteBuf content = httpContent.content(); if (content.isReadable()) { String chunk_message=content.toString(CharsetUtil.UTF_8); buf.append(chunk_message); } if (!(msg instanceof LastHttpContent)) {//<span style="color:#FF0000;">不是最后一个chunk包块,此项非常 重要,当http分多片传输时,需要将多块内容合并</span> return "0"; }else{ decode_message= buf.toString(); // logger.info(decode_message); } if (msg instanceof LastHttpContent) { //LastHttpContent trailer = (LastHttpContent) msg; String sessionId=ctx.channel().id().asLongText(); System.out.println("请求"+decode_message); System.out.println("请求参数"+httpReqParams); System.out.println("请求地址"+uri); System.out.println("会话Id"+sessionId); //<span style="color:#FF0000;">TODO:模拟发送请求消息给后端处理,可以放入消息队列,ctx对象进入会话保持(Map对象)中,等消息队列中处理完成后,恢复会话,并完成消息应答。</span> } } return result_code; } private static AtomicInteger counter = new AtomicInteger(0); @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { manager.getSession().remove(ctx.channel().id().asLongText()); cause.printStackTrace(); ctx.close(); } @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { logger.error("HttpProxyServerHandler[userEventTriggered] "); if (evt instanceof IdleStateEvent) { IdleStateEvent e = (IdleStateEvent) evt; if (e.state() == IdleState.ALL_IDLE) { logger.error("ALL_IDLE"); } else if (e.state() == IdleState.READER_IDLE){ logger.error("READER_IDLE"); }else if (e.state() == IdleState.WRITER_IDLE){ logger.error("WRITER_IDLE"); } ctx.close(); }else if(evt instanceof ErrorEvent){ logger.error(((ErrorEvent) evt).getErrorCode()); ctx.close(); } } }
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelId; /** * http请求会话 * @author wanghao * */ public class SessionContext { private ChannelId channelId =null ;/**会话 channel ID*/ private ChannelHandlerContext ctx;/**会话*/ private String sessionActiveTime;/**会话创建时间*/ private String sessionUnActiveTime;/**会话失效时间*/ public String getSessionActiveTime() { return sessionActiveTime; } public void setSessionActiveTime(String sessionActiveTime) { this.sessionActiveTime = sessionActiveTime; } public String getSessionUnActiveTime() { return sessionUnActiveTime; } public void setSessionunActiveTime(String sessionUnActiveTime) { this.sessionUnActiveTime = sessionUnActiveTime; } public ChannelHandlerContext getCtx() { return ctx; } public void setCtx(ChannelHandlerContext ctx) { this.ctx = ctx; } public ChannelId getChannelId() { return channelId; } public void setChannelId(ChannelId channelId) { this.channelId = channelId; } }
http请求会话管理器
import java.util.Map; /** * http请求会话管理器 * @author * */ public interface SessionContextManager { abstract Map<String,SessionContext> getSession(); }
原文地址:http://blog.csdn.net/wh0426/article/details/45749305