Netty实现java多线程Post请求解析(Map参数类型)—SKY

netty解析Post的键值对

解析时必须加上一个方法，ch.pipeline().addLast(new HttpObjectAggregator(2048)); 放在自己的Handel前面。

http服务器把HttpObjectAggregator放入管道里。HttpObjectAggregator会把多个消息转换为一个单一的FullHttpRequest或是FullHttpResponse。

To solve the problem you either need to offer() all chunks (HttpContent) of a message to HttpPostRequestDecoder before calling getBodyHttpDatas(), or alternatively you can just add the HttpObjectAggregator handler right before your handler to the channel's pipeline. If you do so, HttpObjectAggregator will collect all chunks for you and produce a single FullHttpRequest in place of multiple chunks. Passing FullHttpRequest instead of an ordinary HttpRequest to HttpPostRequestDecoder's constructor eliminates need to offer() chunks.

So you just need to pipeline.addLast(new HttpObjectAggregator(1048576)) before adding your handler. For example:

public class YourServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(1048576));
        pipeline.addLast(new YourServerHandler());
    }
}

一,服务端的编码。

/**
 * Created with IntelliJ IDEA.
 * User: xiaoyongyong
 */
public class HttpServer implements Runnable{
    private final static org.slf4j.Logger LOG = LoggerFactory.getLogger(HttpServer.class);
    private final int port;
    private volatile boolean closed = false;
    private volatile EventLoopGroup bossGroup;
    private volatile EventLoopGroup workerGroup;
    private volatile ServerBootstrap bootstrap;

    public HttpServer(int port) {
        this.port = port;
    }

    public void init() {
        closed = false;
        //配置服务端的NIO线程组
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup);

        bootstrap.channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        // server端发送的是httpResponse，所以要使用HttpResponseEncoder进行编码
                        ch.pipeline().addLast(new HttpResponseEncoder());
                        // server端接收到的是httpRequest，所以要使用HttpRequestDecoder进行解码
                        ch.pipeline().addLast(new HttpRequestDecoder());
                        // 把多个消息转换为一个单一的FullHttpRequest或是FullHttpResponse,解决Post请求参数解析
                        ch.pipeline().addLast(new HttpObjectAggregator(2048));
                        ch.pipeline().addLast(new HttpServerHandler());
                    }
                }).option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true);
    }
    public void bind() throws Exception{
        if (isClosed()) {
            return;
        }
        //绑定端口，开始绑定server，通过调用sync同步方法阻塞直到绑定成功
        ChannelFuture channelFuture = bootstrap.bind(port).sync();
        System.out.println("HTTP服务器启动成功，端口号：" + port);
        //应用程序会一直等待，直到channel关闭
        channelFuture.channel().closeFuture().sync();
        System.out.println("服务器退出完毕，端口号：" + port);
    }
    public void close() {
        closed = true;
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        System.out.println("关闭http服务器: " + port);
    }
    @Override
    public void run() {
        try {
            HttpServer server = new HttpServer(port);
            server.init();
            while (true) {
                try {
                    server.bind();
                }catch (Exception e){
                    LOG.error("",e);
                    e.printStackTrace();
                }finally {
                    server.close();
                }
                Thread.sleep(2000);
            }
        } catch (Exception e) {
            LOG.error("",e);
            e.printStackTrace();
        }
    }
    public boolean isClosed() {
        return closed;
    }
}

二,自己定义的处理类。

/**
 * Created on 2016/7/23.
 *
 * http请求处理类r
 * @author : xiaoyongyong
 */
public class HttpServerHandler extends ChannelHandlerAdapter {

    private static final Log LOG = LogFactory.getLog(HttpServerHandler.class);
    private static ScheduledExecutorService executor = null;
    private static volatile BlockingQueue<QueueBean> queue = null;
    private static volatile String serverIp = null;
    static {
        executor = Executors.newScheduledThreadPool(Integer.valueOf(PropertiesUtil.readValue(Constants.SERVER_THREAD_POOL)));
        queue = new ArrayBlockingQueue<>(Integer.valueOf(PropertiesUtil.readValue(Constants.SERVER_QUEUE_CAPACITY)));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (serverIp == null) {
            serverIp = ((InetSocketAddress) ctx.channel().localAddress()).getAddress().getHostAddress();
        }
        queue.put(new QueueBean(ctx, msg));
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        LOG.error("ctx close!",cause);
        ctx.close();
    }

    class QueueBean {
        private final ChannelHandlerContext ctx;
        private final Object msg;
        QueueBean(ChannelHandlerContext ctx, Object msg) {
            this.ctx = ctx;
            this.msg = msg;
        }
        public ChannelHandlerContext getCtx() {
            return ctx;
        }
        public Object getMsg() {
            return msg;
        }
    }

    static {
        Thread thread = new Thread(new ChannelReadScan(),"ServerChannelReadScan");
        thread.setDaemon(true);
        thread.start();
    }

    private static class ChannelReadScan implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    final QueueBean queueBean = queue.take();
                    executor.execute(new Runnable() {
                        @Override
                        public void run() {
                            try{
                                ChannelHandlerContext ctx = queueBean.getCtx();
                                Object msg = queueBean.getMsg();
                                if (msg instanceof HttpRequest) {

                                    HttpRequest req = (HttpRequest) msg;
                                    if (HttpHeaders.is100ContinueExpected(req)) {
                                        ctx.write(new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.CONTINUE));
                                    }
                                    boolean keepAlive = HttpHeaders.isKeepAlive(req);

                                    // 解析http头部
                                    for (Map.Entry<String, String> h : req.headers()) {
                                        LOG.debug("HEADER: " + h.getKey() + " = " + h.getValue());
                                    }
                                    String uri = req.getUri();
                                    LOG.debug("uri:" + uri);
                                    if (uri.endsWith("/favicon.ico")) {
                                        return;
                                    }
                                    if (uri.startsWith("http")) {
                                        uri = uri.replaceAll("http://[^/]+","");
                                    }
                                    String requestPath = uri.trim().split("\?")[0];

                                    Map<String, String> params = convertToMap(uri,req);

                                    Object result = service(requestPath, params,req,ctx);

                                    FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(JSON.toJSONString(result).getBytes()));
                                    response.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
                                    response.headers().set(CONTENT_LENGTH, response.content().readableBytes());

                                    if (!keepAlive) {
                                        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
                                    } else {
                                        response.headers().set(CONNECTION, Values.KEEP_ALIVE);
                                        ctx.writeAndFlush(response);
                                    }
                                }
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }
                    });
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        private Map<String, String> convertToMap(String uri,HttpRequest req) {
            Map<String, String> params = new HashMap<>();

            // 是GET请求
            if (HttpMethod.GET.equals(req.getMethod())) {
                // 解析请求参数
                QueryStringDecoder queryStringDecoder = new QueryStringDecoder(uri);
                Map<String, List<String>> paramMap = queryStringDecoder.parameters();
                for (Map.Entry<String, List<String>> entry : paramMap.entrySet()) {
                    params.put(entry.getKey(), entry.getValue().get(0));
                }
            }

            if (HttpMethod.POST.equals(req.getMethod())) {
                // 是POST请求
                HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(new DefaultHttpDataFactory(false), req);
                List<InterfaceHttpData> postList = decoder.getBodyHttpDatas();
                for (InterfaceHttpData data : postList) {
                    if (data.getHttpDataType() == InterfaceHttpData.HttpDataType.Attribute) {
                        MemoryAttribute attribute = (MemoryAttribute) data;
                        params.put(attribute.getName(), attribute.getValue());
                    }
                }
            }
            return params;
        }

        private Object service(String requestPath, Map<String,String> params,HttpRequest req,ChannelHandlerContext ctx) {
            long startTime = System.currentTimeMillis();
            Object result;
            try {
                result = Dispatcher.service(requestPath, params);
                long spentTime = System.currentTimeMillis() - startTime;
                log(requestPath, req, ctx, (List<SyncMonitor>) result, spentTime);
            } catch (Exception e) {
                e.printStackTrace();
                result = SyncMonitor.createExceptionResult("服务器异常:" + e.getCause().getMessage());
            }
            return result;
        }

        private void log(String requestPath, HttpRequest req, ChannelHandlerContext ctx,List<SyncMonitor> result, long spentTime) {
            for(SyncMonitor syncMonitor: result){
                LogBean logBean = new LogBean();
                logBean.setAction(requestPath);
                logBean.setIn_param(requestPath);
                logBean.setC_ip(getClientIp(req, ctx));
                logBean.setS_ip(serverIp);
                int status = (syncMonitor.getStatus());
                if (status == SyncMonitor.STATUS_EXCEPTION) {
                    logBean.setError_msg(JSON.toJSONString(syncMonitor.getMessage()));
                }
                logBean.setError_no(status + "");
                logBean.setResult(status + "");
                logBean.setSpent_time(spentTime + "");
                logBean.setLog_type("info");
                logBean.setSys_no("trade_data_monitor");
                LogUtil.info(logBean);
            }
        }

        private String getClientIp(HttpRequest req,ChannelHandlerContext ctx) {
            String clientIP = req.headers().get("X-Forwarded-For");
            if (clientIP == null) {
                InetSocketAddress insocket = (InetSocketAddress) ctx.channel()
                        .remoteAddress();
                clientIP = insocket.getAddress().getHostAddress();
            }
            return clientIP;
        }
    }
}

三，客户端请求。

/**
 * Created on 2015/12/19.
 * @author   xiaoyongyong
 */
public class HttpClientUtil {
    public static final String METHOD_POST = "post";
    public static final String METHOD_GET = "get";
    public static CloseableHttpClient httpclient;
    public static  CloseableHttpClient getHttpClient() {
        if (httpclient == null) {
            PoolingHttpClientConnectionManager connManager = new PoolingHttpClientConnectionManager();
            // Configure total max or per route limits for persistent connections
            // that can be kept in the pool or leased by the connection manager.
            connManager.setMaxTotal(100);
            connManager.setDefaultMaxPerRoute(10);
            httpclient = HttpClients.custom().setConnectionManager(connManager).build();
        }
        return httpclient;
    }

    /**
     * 发送请求
     * @throws Exception
     */
    public static byte[] sendRequest(String url,String methodType,boolean isDataBack) throws Exception {
        CloseableHttpClient httpclient = getHttpClient();//HttpClients.createDefault();
        HttpResponse response = null;
        HttpUriRequest httpRequest = null;
        byte[] data = null;
        try {
            url = URLEncoder.encode(url, "UTF-8");
            if (METHOD_GET.equals(methodType)) {
                httpRequest = new HttpPost(url);
            } else if (METHOD_POST.equals(methodType)) {
                httpRequest = new HttpGet(url);
            }
//            System.out.println("Executing request " + httpRequest.getRequestLine());
//            // Create a custom response handler
            ResponseHandler<String> responseHandler = new ResponseHandler<String>() {
                @Override
                public String handleResponse(
                        final HttpResponse response) throws IOException {
                    int status = response.getStatusLine().getStatusCode();
                    System.out.println(status + ":" + response.getStatusLine().getReasonPhrase());
                    if (status >= 200 && status < 300) {
                        HttpEntity entity = response.getEntity();
                        return entity != null ? EntityUtils.toString(entity) : null;
                    } else {
                        throw new ClientProtocolException("Unexpected response status: " + status);
                    }
                }
            };
            if (isDataBack) {
                data = parseHttpEntity(response.getEntity());
            }
            String responseBody = httpclient.execute(httpRequest, responseHandler);
            System.out.println("responseBody:"+responseBody);
        } finally {
            if (response != null) {
                EntityUtils.consume(response.getEntity());
            }
//            httpclient.close();
        }
        return data;
    }

    /**
     * 解析httpEntity
     * @throws Exception
     */
    public static byte[] parseHttpEntity(HttpEntity entity) throws Exception {
        try (InputStream in = entity.getContent()) {
            return IOUtils.toByteArray(in);
        }
    }


    /**
     * Post传递键值对参数
     */
    public static String sendPostRequest(String url, Map<String, String> param) throws Exception{
        List<BasicNameValuePair> params = new ArrayList<>();
        for (Map.Entry<String ,String> entry :param.entrySet()){
            params.add(new BasicNameValuePair(entry.getKey(),entry.getValue()));
        }
        HttpPost httpPost = new HttpPost(url);
        httpPost.setEntity(new UrlEncodedFormEntity(params));
        return sendPostRequest(httpPost);
    }

    /**
     * 发送请求
     */
    public static String sendPostRequest(HttpUriRequest httpRequest) throws Exception {
        CloseableHttpClient httpClient = getHttpClient();//HttpClients.createDefault();
        HttpResponse response = null;
        String responseBody = "";
        try {
            ResponseHandler<String> responseHandler = new ResponseHandler<String>() {
                @Override
                public String handleResponse(final HttpResponse response) throws IOException {
                    int status = response.getStatusLine().getStatusCode();
                    if (status >= 200 && status < 300) {
                        HttpEntity entity = response.getEntity();
                        return entity != null ? EntityUtils.toString(entity) : null;
                    } else {
                        throw new ClientProtocolException("Unexpected response status: " + status);
                    }
                }
            };
            responseBody = httpClient.execute(httpRequest, responseHandler);
        }catch (Exception e){
            e.printStackTrace();
            throw e;
        }finally {
            if (response != null) {
                EntityUtils.consume(response.getEntity());
            }
        }
        return responseBody;
    }

}