Java并发（二）

public class Lock{
	private boolean isLocked = false;

	public synchronized void lock()
		throws InterruptedException{
		while(isLocked){ //自旋锁防止虚假唤醒
			wait();
		}
		isLocked = true;
	}

	public synchronized void unlock(){
		isLocked = false;
		notify();
	}
}

public class Counter{
	private Lock lock = new Lock();
	private int count = 0;

	public int inc(){
		lock.lock();
		int newCount = ++count;
		lock.unlock();
		return newCount;
	}
}

public class Reentrant2{
	Lock lock = new Lock();

	public outer(){
		lock.lock();
		inner();
		lock.unlock();
	}

	public synchronized inner(){
		lock.lock();
		//do something
		lock.unlock();
	}
}

public class Lock{
	boolean isLocked = false;
	Thread  lockedBy = null;
	int lockedCount = 0;

	public synchronized void lock()
		throws InterruptedException{
		Thread callingThread =
			Thread.currentThread();
		while(isLocked && lockedBy != callingThread){
			wait();
		}
		isLocked = true;
		lockedCount++;
		lockedBy = callingThread;
  }

	public synchronized void unlock(){
		if(Thread.curentThread() ==
			this.lockedBy){
			lockedCount--;

			if(lockedCount == 0){
				isLocked = false;
				notify();
			}
		}
	}

	...
}

 class BoundedBuffer {
   final Lock lock = new ReentrantLock();
   final Condition notFull  = lock.newCondition();
   final Condition notEmpty = lock.newCondition();

   final Object[] items = new Object[100];
   int putptr, takeptr, count;

   public void put(Object x) throws InterruptedException {
     lock.lock();
     try {
       while (count == items.length)
         notFull.await();
       items[putptr] = x;
       if (++putptr == items.length) putptr = 0;
       ++count;
       notEmpty.signal();
     } finally {
       lock.unlock();
     }
   }

   public Object take() throws InterruptedException {
     lock.lock();
     try {
       while (count == 0)
         notEmpty.await();
       Object x = items[takeptr];
       if (++takeptr == items.length) takeptr = 0;
       --count;
       notFull.signal();
       return x;
     } finally {
       lock.unlock();
     }
   }
 }

import java.util.concurrent.*;

public class Main{

	//计算[start, end]范围内整数的和
	public static class CountTask extends RecursiveTask<Integer>{
		private static final int THRESHOLD = 2;
		private int start;
		private int end;

		public CountTask(int start, int end) {
			this.start = start;
			this.end = end;
		}

		@Override
		public String toString(){
			return "CountTast("+start+", "+end+")";
		}


		@Override
		protected Integer compute() {
			System.out.println(this.toString()+" Work in "+Thread.currentThread().getName());
			int sum = 0;
			boolean canCompute = (end - start) <= THRESHOLD;
			if (canCompute) {
				for (int i = start; i <= end; i++)
					sum += i;
			} else {
				//如果任务大于阀值，就分裂成两个子任务计算
				int mid = (start + end) / 2;
				CountTask leftTask = new CountTask(start, mid);
				CountTask rightTask = new CountTask(mid+1, end);

				//执行子任务
				leftTask.fork();
				rightTask.fork();

				//等待子任务执行完，并得到结果
				int leftResult = (int)leftTask.join();
				int rightResult = (int)rightTask.join();

				sum = leftResult + rightResult;
			}

			return sum;
		}
	}

	public static void main(String[] args) throws Exception{
		ForkJoinPool forkJoinPool = new ForkJoinPool(); //创建一个ForkJoin线程池
		 //生成一个计算资格，负责计算1+2+3+4
		CountTask task = new CountTask(1, 4);
		Future<Integer> result = forkJoinPool.submit(task);
		Thread.sleep(1000);
		if(result.isDone()){
			System.out.println(result.get());
		}
	}
}

Result solve(Problem problem) {
	if (problem足够小)
		直接解决问题
	else {
			将问题分割成独立的部分
			衍生出新的子任务无解决每一部分（并行、使用独立的线程，与普通递归算法不同的地方）
			等待所有的子任务完成
			根据子任务结果得到问题的结果并返回
		}
}

public final ForkJoinTask fork() {
        ((ForkJoinWorkerThread) Thread.currentThread())
            .pushTask(this);
        return this;
}

import java.io.*;
import java.nio.*;
import java.nio.channels.*;

public class Main{
	public static void main(String[] args) throws Exception{
		RandomAccessFile aFile = new RandomAccessFile("nio-data.txt", "rw");
		//获得文件频道
		FileChannel inChannel = aFile.getChannel();
		//字节缓冲
		ByteBuffer buf = ByteBuffer.allocate(48);
		//异步读到Buffer中
		int bytesRead = inChannel.read(buf);
		while (bytesRead != -1) {
			//输出上次读了多长
			System.out.println("Read " + bytesRead);
			buf.flip(); //反转Buffer

			while(buf.hasRemaining()){ //缓冲区是否有数据
				System.out.print((char) buf.get()); //一个一个读出来
			}

			buf.clear(); //清空缓冲
			bytesRead = inChannel.read(buf); //再读
		}
		aFile.close();
	}
}

ByteBuffer header = ByteBuffer.allocate(128);
ByteBuffer body   = ByteBuffer.allocate(1024);

ByteBuffer[] bufferArray = { header, body };

channel.read(bufferArray);

ByteBuffer header = ByteBuffer.allocate(128);
ByteBuffer body   = ByteBuffer.allocate(1024);

//write data into buffers

ByteBuffer[] bufferArray = { header, body };

channel.write(bufferArray);

RandomAccessFile fromFile = new RandomAccessFile("fromFile.txt", "rw");
FileChannel      fromChannel = fromFile.getChannel();

RandomAccessFile toFile = new RandomAccessFile("toFile.txt", "rw");
FileChannel      toChannel = toFile.getChannel();

long position = 0;
long count = fromChannel.size();

toChannel.transferFrom(position, count, fromChannel);

Selector selector = Selector.open();

channel.configureBlocking(false);
SelectionKey key = channel.register(selector,
	Selectionkey.OP_READ);

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;

Set selectedKeys = selector.selectedKeys()

Set selectedKeys = selector.selectedKeys();
Iterator keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
    SelectionKey key = keyIterator.next();
    if(key.isAcceptable()) {
        // a connection was accepted by a ServerSocketChannel.
    } else if (key.isConnectable()) {
        // a connection was established with a remote server.
    } else if (key.isReadable()) {
        // a channel is ready for reading
    } else if (key.isWritable()) {
        // a channel is ready for writing
    }
    keyIterator.remove();
}

RandomAccessFile aFile = new RandomAccessFile("data/nio-data.txt", "rw");
FileChannel inChannel = aFile.getChannel();

String newData = "New String to write to file..." + System.currentTimeMillis();

ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());

buf.flip();

while(buf.hasRemaining()) {
	channel.write(buf);
}

SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("www.baidu.com", 80));

		ByteBuffer buf = ByteBuffer.allocate(2048);
		//切换到写模式
		buf.clear();
		buf.put(("GET / HTTP/1.1\r\n"+
			"Host: www.baidu.com\r\n"+
			"Connection: keep-alive\r\n\r\n").getBytes());
		//切换到读模式
		buf.flip();
		//写入网络
		while(buf.hasRemaining()) {
			socketChannel.write(buf);
		}
		buf.clear();

		buf.clear();
		int len = socketChannel.read(buf);
		while (len != -1) {
			buf.flip(); //反转Buffer
			System.out.println(charset.decode(buf));
			buf.clear(); //清空缓冲
			len = socketChannel.read(buf); //再读
		}

socketChannel.connect(new InetSocketAddress("www.baidu.com", 80));
while(! socketChannel.finishConnect() ){
    //wait, or do something else...
}

import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import java.net.*;
import java.nio.charset.Charset;
import java.util.*;
import java.text.*;

public class Main{
	//编码
	final static Charset charset = Charset.forName("utf-8");

	//时间
	final static SimpleDateFormat sdf = new SimpleDateFormat("EEE, d MMM yyyy HH:mm:ss 'GMT'", Locale.US);
	static{
		sdf.setTimeZone(TimeZone.getTimeZone("GMT"));
	}

	public static void main(String[] args) throws Exception{
		//打开一个服务端套接字通道
		ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
		//绑定本机端口
		serverSocketChannel.socket().bind(new InetSocketAddress(8080));
		//创建缓冲
		ByteBuffer buf = ByteBuffer.allocate(2048);
		//循环接收连接
		while(true){
			SocketChannel socketChannel =
				serverSocketChannel.accept();
			//读数据
			while (socketChannel.read(buf)!=-1) {
				buf.flip(); //反转Buffer
				String req = charset.decode(buf).toString();
				System.out.println(req);
				buf.clear();
				if(req.contains("\r\n\r\n")){
					break;
				}
			}
			//写回数据
			buf.clear();
			String resp = buildBody("<!DOCTYPE html>"+
					"<html lang=\"zh_CN\">"+
					"<head>"+
					"	<meta charset=\"UTF-8\">"+
					"	<title>Test</title>"+
					"</head>"+
					"<body>"+
					"	<h1>Hello World</h1>"+
					"</body>"+
					"</html>");
			buf.put(resp.getBytes());
			buf.flip(); //切换到buffer模式
			System.out.println(resp);
			while(buf.hasRemaining()){
				socketChannel.write(buf);
			}
			buf.clear();
			socketChannel.close();
		}
	}

	private static String buildBody(String body){
		StringBuilder sb = new StringBuilder();
		sb.append("HTTP/1.1 200 OK\r\n");
		sb.append("Content-Type: text/html;charset=utf-8\r\n");
		sb.append("Content-Length: "+ body.getBytes().length +"\r\n");
		sb.append("Content-Encoding:  default\r\n");
		sb.append("Server: localhost\r\n");
		Calendar cd = Calendar.getInstance();
		sb.append("Date: "+ sdf.format(cd.getTime()) +"\r\n");
		sb.append("\r\n");
		sb.append(body);
		return sb.toString();
	}
}

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

serverSocketChannel.socket().bind(new InetSocketAddress(9999));
serverSocketChannel.configureBlocking(false);

while(true){
    SocketChannel socketChannel =
            serverSocketChannel.accept();

    if(socketChannel != null){
        //do something with socketChannel...
    }
}

import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import java.net.*;
import java.nio.charset.Charset;
import java.util.*;
import java.text.*;

public class NonBlockingServer{
	//编码
	final static Charset charset = Charset.forName("utf-8");

	//时间
	final static SimpleDateFormat sdf = new SimpleDateFormat("EEE, d MMM yyyy HH:mm:ss 'GMT'", Locale.US);
	static{
		sdf.setTimeZone(TimeZone.getTimeZone("GMT"));
	}

	public static void main(String[] args) throws Exception{
		//创建多路复用器
		Selector selector = Selector.open();
		//打开一个服务端套接字通道
		ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
		//服务端套接字设为false
		serverSocketChannel.configureBlocking(false);
		//绑定本机端口
		serverSocketChannel.socket().bind(new InetSocketAddress(8080));
		//在复用器中注册服务端socket
		serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
		//循环接收连接
		while(true){
			//阻塞等待套接字，准备就绪
			selector.select();
			//获取准备好的套接字Key的迭代器
			Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
			//遍历
			while(keyIterator.hasNext()){
				SelectionKey key = keyIterator.next();
				if(key.isValid()) handle(key);
				keyIterator.remove();
			}
		}
	}

	private static void handle(SelectionKey key) throws Exception{
		//服务端socket
		if(key.isAcceptable()) {
			//获取该通道
			ServerSocketChannel channel = (ServerSocketChannel) key.channel();
			//获取socket
			SocketChannel socketChannel = channel.accept();
			//设为非阻塞
			socketChannel.configureBlocking(false);
			//注册到selector中
			//注册读
			socketChannel.register(key.selector(), SelectionKey.OP_READ);
		}
		if(key.isReadable() || key.isWritable()) {
			httpService(key);
		}
	}

	private static void httpService(SelectionKey key) throws Exception{
		if(key.isReadable()){
			//创建缓冲
			ByteBuffer buf = ByteBuffer.allocate(2048);
			SocketChannel socketChannel = (SocketChannel) key.channel();
			//读数据
			while (socketChannel.read(buf)!=-1) {
				buf.flip(); //反转Buffer
				String req = charset.decode(buf).toString();
				System.out.println(req);
				buf.clear();
				if(req.contains("\r\n\r\n")){
					break;
				}
			}
			//写回数据
			buf.clear();
			String resp = buildBody("<!DOCTYPE html>"+
					"<html lang=\"zh_CN\">"+
					"<head>"+
					"	<meta charset=\"UTF-8\">"+
					"	<title>Test</title>"+
					"</head>"+
					"<body>"+
					"	<h1>Hello World</h1>"+
					"</body>"+
					"</html>");
			buf.put(resp.getBytes());
			buf.flip(); //切换到buffer模式
			System.out.println(resp);
			while(buf.hasRemaining()){
				socketChannel.write(buf);
			}
			buf.clear();
			// key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
			key.cancel();
			socketChannel.close();
		}
	}

	private static String buildBody(String body){
		StringBuilder sb = new StringBuilder();
		sb.append("HTTP/1.0 200 OK\r\n");
		sb.append("Content-Type: text/html;charset=utf-8\r\n");
		sb.append("Content-Length: "+ body.getBytes().length +"\r\n");
		sb.append("Content-Encoding:  default\r\n");
		sb.append("Server: localhost\r\n");
		Calendar cd = Calendar.getInstance();
		sb.append("Date: "+ sdf.format(cd.getTime()) +"\r\n");
		sb.append("\r\n");
		sb.append(body);
		return sb.toString();
	}
}

DatagramChannel channel = DatagramChannel.open();
channel.socket().bind(new InetSocketAddress(9999));

ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
SocketAddress address = channel.receive(buf);

String newData = "New String to write to file..." + System.currentTimeMillis();

ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();

int bytesSent = channel.send(buf, new InetSocketAddress("jenkov.com", 80));

channel.connect(new InetSocketAddress("jenkov.com", 80));
int bytesRead = channel.read(buf);
int bytesWritten = channel.write(but);

import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import java.net.*;
import java.nio.charset.Charset;
import java.util.*;
import java.text.*;
import java.util.concurrent.*;

public class Main{
	final static Charset charset = Charset.defaultCharset();

	public static void main(String[] args) throws Exception{
		Pipe pipe = Pipe.open();
		ExecutorService es = Executors.newCachedThreadPool();

		es.submit(()->{
			Pipe.SinkChannel sinkChannel = pipe.sink();
			ByteBuffer buf = ByteBuffer.allocate(48);
			boolean isRunning = true;
			while(isRunning){
				buf.clear();
				buf.put(("send from"+Thread.currentThread().getName()).getBytes());
				buf.flip();
				while(buf.hasRemaining()) {
					sinkChannel.write(buf);
				}
				Thread.sleep(1000);
			}
			return 0; //解析成Callable允许抛异常
		});
		es.submit(()->{
			Pipe.SourceChannel sourceChannel = pipe.source();
			ByteBuffer buf = ByteBuffer.allocate(48);
			int len = sourceChannel.read(buf);
			while (len != -1) {
				buf.flip(); //反转Buffer
				System.out.println(Thread.currentThread().getName()+" recevice:");
				System.out.println(charset.decode(buf));
				buf.clear(); //清空缓冲
				len = sourceChannel.read(buf);//再读
			}
			return 0; //解析成Callable允许抛异常
		});
	}
}

Path path=Paths.get("D://test/test.txt");
Path path1=Paths.get("/home/opt/app/test/test.txt"); //若在Windows中解析成C:/home/opt/app/test/test.txt
Path path2=Paths.get("d:\\data", "projects\\a-project\\myfile.txt"); //创建相对路径Paths.get(basePath, relativePath)

Path currentDir = Paths.get("."); //当前目录
System.out.println(currentDir.toAbsolutePath());
Path parentDir = Paths.get(".."); //父目录

Path path = Paths.get("data/test.xml");
AsynchronousFileChannel fileChannel =
    AsynchronousFileChannel.open(path, StandardOpenOption.READ);

Future<Integer> operation = fileChannel.read(buffer, 0);

fileChannel.read(buffer, position, buffer, new CompletionHandler<Integer, ByteBuffer>() {
    @Override
    public void completed(Integer result, ByteBuffer attachment) {
        System.out.println("result = " + result);

        attachment.flip();
        byte[] data = new byte[attachment.limit()];
        attachment.get(data);
        System.out.println(new String(data));
        attachment.clear();
    }

    @Override
    public void failed(Throwable exc, ByteBuffer attachment) {

    }
});

public interface Callable<V> {
    V call() throws Exception;
}

<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning); //取消正在执行的任务
    boolean isCancelled(); //判断是否取消成功
    boolean isDone(); //判断是否执行完毕
    V get() throws InterruptedException, ExecutionException; //获得线程结果
    V get(long timeout, TimeUnit unit)  //获取线程结果
        throws InterruptedException, ExecutionException, TimeoutException;
}