package rwbigfile;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.nio.channels.ReadableByteChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;

import util.StopWatch;

/**
 * NIO寫大文件比較
 * @author Will
 * 
 */
public class WriteBigFileComparison {

	// data chunk be written per time
	private static final int DATA_CHUNK = 128 * 1024 * 1024; 

	// total data size is 2G
	private static final long LEN = 2L * 1024 * 1024 * 1024L; 

	
	public static void writeWithFileChannel() throws IOException {
		File file = new File("e:/test/fc.dat");
		if (file.exists()) {
			file.delete();
		}

		RandomAccessFile raf = new RandomAccessFile(file, "rw");
		FileChannel fileChannel = raf.getChannel();

		byte[] data = null;
		long len = LEN;
		ByteBuffer buf = ByteBuffer.allocate(DATA_CHUNK);
		int dataChunk = DATA_CHUNK / (1024 * 1024);
		while (len >= DATA_CHUNK) {
			System.out.println("write a data chunk: " + dataChunk + "MB");

			buf.clear(); // clear for re-write
			data = new byte[DATA_CHUNK];
			for (int i = 0; i < DATA_CHUNK; i++) {
				buf.put(data[i]);
			}

			data = null;

			buf.flip(); // switches a Buffer from writing mode to reading mode
			fileChannel.write(buf);
			fileChannel.force(true);

			len -= DATA_CHUNK;
		}

		if (len > 0) {
			System.out.println("write rest data chunk: " + len + "B");
			buf = ByteBuffer.allocateDirect((int) len);
			data = new byte[(int) len];
			for (int i = 0; i < len; i++) {
				buf.put(data[i]);
			}

			buf.flip(); // switches a Buffer from writing mode to reading mode, position to 0, limit not changed
			fileChannel.write(buf);
			fileChannel.force(true);
			data = null;
		}

		fileChannel.close();
		raf.close();
	}

	/**
	 * write big file with MappedByteBuffer
	 * @throws IOException
	 */
	public static void writeWithMappedByteBuffer() throws IOException {
		File file = new File("e:/test/mb.dat");
		if (file.exists()) {
			file.delete();
		}

		RandomAccessFile raf = new RandomAccessFile(file, "rw");
		FileChannel fileChannel = raf.getChannel();
		int pos = 0;
		MappedByteBuffer mbb = null;
		byte[] data = null;
		long len = LEN;
		int dataChunk = DATA_CHUNK / (1024 * 1024);
		while (len >= DATA_CHUNK) {
			System.out.println("write a data chunk: " + dataChunk + "MB");

			mbb = fileChannel.map(MapMode.READ_WRITE, pos, DATA_CHUNK);
			data = new byte[DATA_CHUNK];
			mbb.put(data);

			data = null;

			len -= DATA_CHUNK;
			pos += DATA_CHUNK;
		}

		if (len > 0) {
			System.out.println("write rest data chunk: " + len + "B");

			mbb = fileChannel.map(MapMode.READ_WRITE, pos, len);
			data = new byte[(int) len];
			mbb.put(data);
		}

		data = null;
		unmap(mbb);  // release MappedByteBuffer
		fileChannel.close();
	}
	
	public static void writeWithTransferTo() throws IOException {
		File file = new File("e:/test/transfer.dat");
		if (file.exists()) {
			file.delete();
		}
		
		RandomAccessFile raf = new RandomAccessFile(file, "rw");
		FileChannel toFileChannel = raf.getChannel();
		
		long len = LEN;
		byte[] data = null;
		ByteArrayInputStream bais = null;
		ReadableByteChannel fromByteChannel = null;
		long position = 0;
		int dataChunk = DATA_CHUNK / (1024 * 1024);
		while (len >= DATA_CHUNK) {
			System.out.println("write a data chunk: " + dataChunk + "MB");
			
			data = new byte[DATA_CHUNK];
			bais = new ByteArrayInputStream(data);
			fromByteChannel = Channels.newChannel(bais);
			
			long count = DATA_CHUNK;
			toFileChannel.transferFrom(fromByteChannel, position, count);
			
			data = null;
			position += DATA_CHUNK;
			len -= DATA_CHUNK;
		}
		
		if (len > 0) {
			System.out.println("write rest data chunk: " + len + "B");

			data = new byte[(int) len];
			bais = new ByteArrayInputStream(data);
			fromByteChannel = Channels.newChannel(bais);
			
			long count = len;
			toFileChannel.transferFrom(fromByteChannel, position, count);
		}
		
		data = null;
		toFileChannel.close();
		fromByteChannel.close();
	}
	
	/**
	 * 在MappedByteBuffer釋放后再對它進(jìn)行讀操作的話就會(huì)引發(fā)jvm crash，在并發(fā)情況下很容易發(fā)生
	 * 正在釋放時(shí)另一個(gè)線程正開始讀取，于是crash就發(fā)生了。所以為了系統(tǒng)穩(wěn)定性釋放前一般需要檢
	 * 查是否還有線程在讀或?qū)?
	 * @param mappedByteBuffer
	 */
	public static void unmap(final MappedByteBuffer mappedByteBuffer) {
		try {
			if (mappedByteBuffer == null) {
				return;
			}
			
			mappedByteBuffer.force();
			AccessController.doPrivileged(new PrivilegedAction<Object>() {
				@Override
				@SuppressWarnings("restriction")
				public Object run() {
					try {
						Method getCleanerMethod = mappedByteBuffer.getClass()
								.getMethod("cleaner", new Class[0]);
						getCleanerMethod.setAccessible(true);
						sun.misc.Cleaner cleaner = 
								(sun.misc.Cleaner) getCleanerMethod
									.invoke(mappedByteBuffer, new Object[0]);
						cleaner.clean();
						
					} catch (Exception e) {
						e.printStackTrace();
					}
					System.out.println("clean MappedByteBuffer completed");
					return null;
				}
			});

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) throws IOException {
		StopWatch sw = new StopWatch();
		
		sw.startWithTaskName("write with file channel's write(ByteBuffer)");
		writeWithFileChannel();
		sw.stopAndPrint();
		
		sw.startWithTaskName("write with file channel's transferTo");
		writeWithTransferTo();
		sw.stopAndPrint();
		
		sw.startWithTaskName("write with MappedByteBuffer");
		writeWithMappedByteBuffer();
		sw.stopAndPrint();
	}

}

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;

public class MappedFile {
	
	// 文件名
	private String fileName;

	// 文件所在目錄路徑
	private String fileDirPath;

	// 文件對象
	private File file;

	private MappedByteBuffer mappedByteBuffer;
	private FileChannel fileChannel;
	private boolean boundSuccess = false;

	// 文件最大只能為50MB
	private final static long MAX_FILE_SIZE = 1024 * 1024 * 50;
	
	// 最大的臟數(shù)據(jù)量512KB,系統(tǒng)必須觸發(fā)一次強(qiáng)制刷
	private long MAX_FLUSH_DATA_SIZE = 1024 * 512;

	// 最大的刷間隔,系統(tǒng)必須觸發(fā)一次強(qiáng)制刷
	private long MAX_FLUSH_TIME_GAP = 1000;

	// 文件寫入位置
	private long writePosition = 0;

	// 最后一次刷數(shù)據(jù)的時(shí)候
	private long lastFlushTime;

	// 上一次刷的文件位置
	private long lastFlushFilePosition = 0;
	
	public MappedFile(String fileName, String fileDirPath) {
		super();
		this.fileName = fileName;
		this.fileDirPath = fileDirPath;
		this.file = new File(fileDirPath + "/" + fileName);
		if (!file.exists()) {
			try {
				file.createNewFile();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		
	}

	/**
	 * 
	 * 內(nèi)存映照文件綁定
	 * @return
	 */
	public synchronized boolean boundChannelToByteBuffer() {
		try {
			RandomAccessFile raf = new RandomAccessFile(file, "rw");
			this.fileChannel = raf.getChannel();
		} catch (Exception e) {
			e.printStackTrace();
			this.boundSuccess = false;
			return false;
		}

		try {
			this.mappedByteBuffer = this.fileChannel
					.map(FileChannel.MapMode.READ_WRITE, 0, MAX_FILE_SIZE);
		} catch (IOException e) {
			e.printStackTrace();
			this.boundSuccess = false;
			return false;
		}

		this.boundSuccess = true;
		return true;
	}
	
	/**
	 * 寫數(shù)據(jù)：先將之前的文件刪除然后重新
	 * @param data
	 * @return
	 */
	public synchronized boolean writeData(byte[] data) {
		
		return false;
	}
	
	/**
	 * 在文件末尾追加數(shù)據(jù)
	 * @param data
	 * @return
	 * @throws Exception
	 */
	public synchronized boolean appendData(byte[] data) throws Exception {
		if (!boundSuccess) {
			boundChannelToByteBuffer();
		}
		
		writePosition = writePosition + data.length;
		if (writePosition >= MAX_FILE_SIZE) {  // 如果寫入data會(huì)超出文件大小限制，不寫入
			flush();
			writePosition = writePosition - data.length;
			System.out.println("File=" 
								+ file.toURI().toString() 
								+ " is written full.");
			System.out.println("already write data length:" 
								+ writePosition
								+ ", max file size=" + MAX_FILE_SIZE);
			return false;
		}

		this.mappedByteBuffer.put(data);

		// 檢查是否需要把內(nèi)存緩沖刷到磁盤
		if ( (writePosition - lastFlushFilePosition > this.MAX_FLUSH_DATA_SIZE)
			 ||
			 (System.currentTimeMillis() - lastFlushTime > this.MAX_FLUSH_TIME_GAP
			 && writePosition > lastFlushFilePosition) ) {
			flush();  // 刷到磁盤
		}
		
		return true;
	}

	public synchronized void flush() {
		this.mappedByteBuffer.force();
		this.lastFlushTime = System.currentTimeMillis();
		this.lastFlushFilePosition = writePosition;
	}

	public long getLastFlushTime() {
		return lastFlushTime;
	}

	public String getFileName() {
		return fileName;
	}

	public String getFileDirPath() {
		return fileDirPath;
	}

	public boolean isBundSuccess() {
		return boundSuccess;
	}

	public File getFile() {
		return file;
	}

	public static long getMaxFileSize() {
		return MAX_FILE_SIZE;
	}

	public long getWritePosition() {
		return writePosition;
	}

	public long getLastFlushFilePosition() {
		return lastFlushFilePosition;
	}

	public long getMAX_FLUSH_DATA_SIZE() {
		return MAX_FLUSH_DATA_SIZE;
	}

	public long getMAX_FLUSH_TIME_GAP() {
		return MAX_FLUSH_TIME_GAP;
	}

}