netty接收大文件的方法

参考:http://blog.csdn.net/linuu/article/details/51371595

https://www.jianshu.com/p/a0a51fd79f62

netty默认是只能接收1024个字节，但是我们要传输大文件怎么办？

上代码：

改之后服务端：

package server;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelPipeline;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

import io.netty.handler.codec.DelimiterBasedFrameDecoder;

import io.netty.handler.codec.string.StringDecoder;

import io.netty.handler.codec.string.StringEncoder;

import java.nio.charset.Charset;

import org.apache.log4j.Logger;

import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.eshore.ismp.hbinterface.service.BizCommonService;

import com.eshore.ismp.hbinterface.util.ConfigLoadUtil;

public class SpsServer {

	 private static final Logger logger = Logger.getLogger(SpsServer.class);

	    private static int PORT = 10001;

	    /**用于分配处理业务线程的线程组个数 */

	    protected static final int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors()*2; //默认

	    /** 业务出现线程大小*/

	    protected static final int BIZTHREADSIZE = 4;

	        /*

	     * NioEventLoopGroup实际上就是个线程池,

	     * NioEventLoopGroup在后台启动了n个NioEventLoop来处理Channel事件,

	     * 每一个NioEventLoop负责处理m个Channel,

	     * NioEventLoopGroup从NioEventLoop数组里挨个取出NioEventLoop来处理Channel

	     */

	    private static final EventLoopGroup bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);

	    private static final EventLoopGroup workerGroup = new NioEventLoopGroup(BIZTHREADSIZE);

	    protected static void run(final BizCommonService bizCommonService) throws Exception {

	    	String PORTs=ConfigLoadUtil.getValue("toSpsServerPort");

	    	PORT=Integer.parseInt(PORTs);

	    	logger.info("PORT IS:"+PORT);

	        ServerBootstrap b = new ServerBootstrap();

	        b.group(bossGroup, workerGroup);

	        b.channel(NioServerSocketChannel.class);

	        b.childHandler(new ChannelInitializer<SocketChannel>() {

	            @Override

	            public void initChannel(SocketChannel ch) throws Exception {

	                ChannelPipeline pipeline = ch.pipeline();

	               /* pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));

	                pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));  */

	                ByteBuf delimiter = Unpooled.copiedBuffer("\t".getBytes());

					pipeline.addLast("framer", new DelimiterBasedFrameDecoder(2048,delimiter));

	                pipeline.addLast("decoder", new StringDecoder(Charset.forName("GBK")));

	                pipeline.addLast("encoder", new StringEncoder(Charset.forName("GBK")));

	                pipeline.addLast(new SpsServerHandler(bizCommonService));

	            }

	        });

	        b.bind(PORT).sync();

	        logger.info("TCP服务器已启动");

	    }

	    protected static void shutdown() {

	        workerGroup.shutdownGracefully();

	        bossGroup.shutdownGracefully();

	    }

	    public static void main(String[] args) throws Exception {

	    	try{

	    		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(

	    				new String[] { "applicationContext.xml" });

	    		context.start();

	    		BizCommonService bizCommonService = (BizCommonService) context.getBean("bizCommonService");

	    		SpsServer.run(bizCommonService);

	    	}catch(Exception e){

	    		logger.error("start sps interface server error:",e);

	    		System.exit(-1);

	    	}

	    }

}

改之后客户端：

package fourNoBlocking;

import java.io.IOException;

import java.io.InputStream;

import java.io.OutputStream;

import java.net.InetSocketAddress;

import java.net.Socket;

import java.nio.ByteBuffer;

/**

 *

 * 发送报文给客户端

 *

 *

 * @date 2016年12月14日 上午11:56:27

 * @since 1.0

 */

public class SendClient {

    private static final String ENCODING = "GBK";

    public static String send(String ip, int port, String sendStr, int timeout) {

        long start = System.currentTimeMillis();

        System.out.println(sendStr.length());

        if (sendStr == null || "".equals(sendStr)) {

            return "str is null";

        }

        Socket client = null;

        OutputStream stream = null;

        InputStream is = null;

        try {

            client = new Socket();

            InetSocketAddress address = new InetSocketAddress(ip, port);

            client.connect(address);

            timeout = timeout >= 0 ? timeout : 3500;

            client.setSoTimeout(timeout);

            stream = client.getOutputStream();

            is = client.getInputStream();

            int len = 0;

            len = sendStr.getBytes(ENCODING).length;

            ByteBuffer buf = ByteBuffer.allocate(len);

            byte[] bytes = sendStr.getBytes(ENCODING);

            buf.put(bytes);

            stream.write(buf.array(), 0, len);

            stream.flush();

            String res = "";

            int i = 0;

            byte[] b = new byte[6555];

            while ((i = is.read(b)) != -1) {

                res = new String(b, 0, i);

                System.out.println(res);

                break;

            }

            long end = System.currentTimeMillis();

            return res;

        } catch (Exception e) {

            StringBuilder strBuilder = new StringBuilder();

            strBuilder.append("error send message").append(e.getMessage()).append("&errorID=")

            .append(System.currentTimeMillis());

            return strBuilder.toString();

        } finally {

            if (client != null) {

                try {

                    client.close();

                } catch (IOException e) {

                    StringBuilder strBuilder = new StringBuilder();

                    strBuilder.append("error send message").append(e.getMessage()).append("&errorID=")

                    .append(System.currentTimeMillis());

                }

            }

            if (stream != null) {

                try {

                    stream.close();

                } catch (IOException e) {

                    StringBuilder strBuilder = new StringBuilder();

                    strBuilder.append("error send message").append(e.getMessage()).append("&errorID=")

                    .append(System.currentTimeMillis());

                }

            }

            if (is != null) {

                try {

                    is.close();

                } catch (IOException e) {

                    StringBuilder strBuilder = new StringBuilder();

                    strBuilder.append("error send message").append(e.getMessage()).append("&errorID=")

                    .append(System.currentTimeMillis());

                }

            }

        }

    }

    public static void main(String[] args) {

    	String msg="";

        msg="FFFF76623634010100102700170103IBSS017555      000000021800100023402287248808*766236340100200001178400003001785000030217860000302110000004075510100020SZ2000000054121442461020001241324186148310300593PM_DJDHHM||83456517||001#$PM_HYLX||0||001#$BA_MSMAN||海豚||001#$PM_DJQYYB||518000||001#$PM_DJQYMC||深圳市福田区人力资源服务中心||001#$PM_BHHM||83456517||001#$PM_DJQYDZ||福田区福强路深圳文化创意园世纪工艺品文化广场309栋B座1-3层||001#$PM_SFZDXY||XY02||001#$PM_DJKHXX||||001#$BA_MSDEPTNAME||12||001#$PM_DLS||DSL6||001#$PM_YWSLLB||SLLB01||001#$PM_SLDYSLSH||0||001#$PM_JFQ||01||001#$PM_DJHMGS||1||001#$PM_SRFJ||2||001#$PM_JFJG||1||001#$PM_YZ||30||001#$PM_DXFSSL||100||001#$PB_BILLINGTYPE||000000||005#$PB_USERTYPE||100002||005#$PB_USERCHAR||JFSX01||005#$BEGIN_DATE||20170607||005#$END_DATE||||005#$10400014DXMP214688722910700016号百信息服务中心10800010122810070411400006徐冬生115000088291816511600110114+企业名片行业版包月套餐,114+短信名片包月套餐_定价计划,114+企业名片行业版包月套餐赠送3个月套餐外等额话费优惠11700017755KH000293285120\t";

    	String x=SendClient.send("127.0.0.1", 10001, msg, 3500);

    	System.out.println("return string:"+x);

	}

}

处理类：

package server;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.SimpleChannelInboundHandler;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import com.eshore.ismp.hbinterface.service.BizCommonService;

public class SpsServerHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger logger = LoggerFactory.getLogger(SpsServerHandler.class);

    private BizCommonService bizCommonService;

    public SpsServerHandler(){}

    public SpsServerHandler(BizCommonService bizCommonService){

    	this.bizCommonService=bizCommonService;

    }

    @Override

    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {

        logger.info("SERVER接收到消息 msg:{}",msg);

        long start = System.currentTimeMillis();

        boolean result = bizCommonService.sendOperToCacheAysn(String.valueOf(msg));

    	/**

    	 * step 3 : 创建响应报文

    	 */

		String res = bizCommonService.createResponseStr(String.valueOf(msg),result);

		long end = System.currentTimeMillis();

        logger.debug("SpsServer request:{} res:{} time cost:{}ms",String.valueOf(msg),res,(end-start));

        ctx.channel().writeAndFlush(res);

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx,

            Throwable cause) throws Exception {

        logger.warn("Unexpected exception from downstream.", cause);

        ctx.close();

    }

	@Override

	public void channelInactive(ChannelHandlerContext ctx) throws Exception {

		logger.info("client closed:"+ctx.channel().hashCode());

		super.channelInactive(ctx);

	}

}

输出：

length:1027

服务端增加了：

ByteBuf delimiter = Unpooled.copiedBuffer("\t".getBytes());

					pipeline.addLast("framer", new DelimiterBasedFrameDecoder(2048,delimiter));

客户端报文增加了

\t