netty实现http api功能

无可致疑，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();
}

}
}

SessionContext

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();
}