Netty文件传输，使用HttpChunk

利用Netty中提供的HttpChunk简单实现文件传输

服务端基本和Netty官方文档中的example：http--file一样。

客户端：

HttpClient：

public class HttpClient {
	private ClientBootstrap bootstrap;
	private String host="localhost";
	private Channel channel;
	private boolean futureSuccess;
	private int port=8080;

	public HttpClient() {
	}

	public ChannelFuture connect() {
		bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors
				.newCachedThreadPool()));
		HttpResponseHandler clientHandler = new HttpResponseHandler();
		bootstrap.setPipelineFactory(new HttpClientPipelineFactory(clientHandler));

		bootstrap.setOption("tcpNoDelay", true);
		bootstrap.setOption("keepAlive", true);

		return bootstrap.connect(new InetSocketAddress(host,port));
	}
	
	public boolean checkFutureState(ChannelFuture channelFuture) {
		// Wait until the connection attempt succeeds or fails.
		channel = channelFuture.awaitUninterruptibly().getChannel();
		channelFuture.addListener(new ChannelFutureListener() {
			@Override
			public void operationComplete(ChannelFuture connectFuture) throws Exception {
				if (!connectFuture.isSuccess()) {
					connectFuture.getCause().printStackTrace();
					// connectFuture.getChannel().close();
					// bootstrap.releaseExternalResources();
					futureSuccess = false;
				} else {
					futureSuccess = true;
				}
			}
		});
		return futureSuccess;
	}
	
	public ChannelFuture write(HttpRequest request) {
		return channel.write(request);
	}
	
	public void Close() {
		// Close the connection. Make sure the close operation ends because
		// all I/O operations are asynchronous in Netty.
		channel.close().awaitUninterruptibly();
		// Shut down all thread pools to exit.
		bootstrap.releaseExternalResources();
	}
}

 HttpClientPipelineFactory：

public class HttpClientPipelineFactory implements ChannelPipelineFactory {
	private final HttpResponseHandler handler;

	public HttpClientPipelineFactory(HttpResponseHandler handler) {
		this.handler = handler;
	}
	
	public ChannelPipeline getPipeline() throws Exception {
		ChannelPipeline pipeline = pipeline();
		
        pipeline.addLast("decoder", new HttpResponseDecoder());
        //pipeline.addLast("aggregator", new HttpChunkAggregator(6048576));
        pipeline.addLast("encoder", new HttpRequestEncoder());
        pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());
        pipeline.addLast("handler", handler);
        
        return pipeline;
	}
}

HttpResponseHandler：

@ChannelPipelineCoverage("one")
public class HttpResponseHandler extends SimpleChannelUpstreamHandler {
	private volatile boolean readingChunks;
	private File downloadFile;
	private FileOutputStream fOutputStream = null;

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
		if (e.getMessage() instanceof HttpResponse) {
			DefaultHttpResponse httpResponse = (DefaultHttpResponse) e.getMessage();
			String fileName = httpResponse.getHeader("Content-Disposition").substring(20);
			downloadFile = new File(System.getProperty("user.dir") + File.separator + "download" + fileName);
			readingChunks = httpResponse.isChunked();
		} else {
			HttpChunk httpChunk = (HttpChunk) e.getMessage();
			if (!httpChunk.isLast()) {
				ChannelBuffer buffer = httpChunk.getContent();
				if (fOutputStream == null) {
					fOutputStream = new FileOutputStream(downloadFile);
				}
				while (buffer.readable()) {
					byte[] dst = new byte[buffer.readableBytes()];
					buffer.readBytes(dst);
					fOutputStream.write(dst);
				}
			} else {
				readingChunks = false;
			}
			fOutputStream.flush();
		}
		if (!readingChunks) {
			fOutputStream.close();
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
		System.out.println(e.getCause());
	}
}

  main:

public class ClientMain {
	public static void main(String[] args) {
		HttpClient httpClient=new HttpClient();
		ChannelFuture connectFuture=httpClient.connect();
		if (httpClient.checkFutureState(connectFuture)) {
			System.out.println("connect ok");
			HttpRequest request=new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "thunder.zip");
//			HttpRequest request=new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "thunder.java");
			ChannelFuture writeFuture= httpClient.write(request);
			if (httpClient.checkFutureState(writeFuture)) {
				System.out.println("write ok");
			}
		}
	}
}

 

改天看看零拷贝方式