NIO&Concurrent

 public final Buffer flip() {
        limit = position;
        position = 0;
        mark = -1;
        return this;
    }

//创建缓冲区并且添加初始元素
		//指定了缓冲区的初始元素以及长度
		ByteBuffer buffer=ByteBuffer.wrap("Hello".getBytes());
		//缓冲区的position现在是多少
		//无论用什么方式创建的缓冲区,默认初始位置都是0
		//System.out.println(buffer.position());
		//buffer.put("a".getBytes());//会覆盖h
		
		byte[] arr=buffer.array();//获取顶层的字节数组
		System.out.println(new String(arr,0,buffer.limit()));

//创建客户端的通道
		SocketChannel s=SocketChannel.open();
		//设置为非阻塞
		s.configureBlocking(false);
		//发链接
		//连接过程在BIO中会产生阻塞
		s.connect(new InetSocketAddress("localhost",8070));
		
		//判断连接是否建立
		while(!s.isConnected()){
			//如果没有连上,试图再次连接
			//如果连接多次依然失败,则意味着这个连接无效
			//finishConnect底层会自动计数,计数多次依然失败,则抛出异常
			s.finishConnect();
		}
		
		//发送数据
		s.write(ByteBuffer.wrap("hello server".getBytes()));
		Thread.sleep(10);
		//接收数据
		ByteBuffer dst=ByteBuffer.allocate(1024);
		s.read(dst);
		System.out.println(new String(dst.array(),0,dst.position()));
		//关流
		s.close();

public class Server {
	public static void main(String[] args) throws IOException, InterruptedException {
		//开启服务器端的通道
		ServerSocketChannel ssc=ServerSocketChannel.open();
		//绑定接收端口
		ssc.bind(new InetSocketAddress(8070));
		//设置非阻塞
		ssc.configureBlocking(false);
		//接收连接
		SocketChannel sc=ssc.accept();
		//判断连接是否接收
		//服务器端是否要计数-不用
		while(sc==null)
			sc=ssc.accept();
		//接收数据
		ByteBuffer buffer = ByteBuffer.allocate(1024);
		sc.read(buffer);
		buffer.flip();
		System.out.println(new String(buffer.array(),0,buffer.limit()));
		//打回数据
		sc.write(ByteBuffer.wrap("数据已经收到".getBytes()));
		Thread.sleep(10);
		//关流
		ssc.close();
	}
}

// 开启服务器端的通道
		ServerSocketChannel ssc = ServerSocketChannel.open();
		// 绑定端口
		ssc.bind(new InetSocketAddress(8070));
		// 将通道设置为非阻塞
		ssc.configureBlocking(false);
		// 开启选择器
		Selector selc = Selector.open();
		// 将服务器注册到选择器上
		ssc.register(selc, SelectionKey.OP_ACCEPT);
		while (true) {
			// 进行选择
			selc.select();
			// 获取选择出来的事件
			Set<SelectionKey> set = selc.selectedKeys();
			// 遍历集合,根据事件不同类型进行处理
			Iterator<SelectionKey> it = set.iterator();
			while (it.hasNext()) {
				SelectionKey key = it.next();
				//需要注意的是,key表示的是一个事件
				// 可接受
				if (key.isAcceptable()) {
					//真正要完成accept操作需要依靠通道来完成
					//从这个事件需要进行accept的通道
					ServerSocketChannel sscx=(ServerSocketChannel) key.channel();
					//接收连接
					SocketChannel sc=sscx.accept();
					//需要给这个通道注册可读或者可写事件
					//如果既需要注册可读,又需要注册可写
					//在注册的时候,后注册的事件会把之前的事件覆盖掉
					//将sc设置为非阻塞
					sc.configureBlocking(false);
					sc.register(selc, SelectionKey.OP_WRITE|SelectionKey.OP_READ);
				}
				// 可读
				if (key.isReadable()) {
					//先从事件中获取通道
					SocketChannel sc=(SocketChannel) key.channel();
					//读数据
					ByteBuffer dst=ByteBuffer.allocate(1024);
					sc.read(dst);
					System.out.println(new String(dst.array(),0,dst.position()));
					//注销read事件
					//sc.register(selc, key.interestOps()-SelectionKey.OP_READ);
					sc.register(selc, key.interestOps()^SelectionKey.OP_READ);
				}
				// 可写
				if (key.isWritable()) {
					//从事件中获取通道
					SocketChannel sc=(SocketChannel)key.channel();
					//写出数据
					sc.write(ByteBuffer.wrap("hello client".getBytes()));
					//注销write事件
					sc.register(selc, key.interestOps()^SelectionKey.OP_WRITE);
				}
				//移除
				it.remove();

        //corePoolSize 核心线程数量
		//maximumPoolSize 最大线程数 =核心线程数+临时线程数
		//keepAliveTime 存活时间
		//unit 单位
		//workQueue 工作队列
		//handler 拒绝执行处理器
			ExecutorService es=new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(5),new RejectedExecutionHandler() {
			
			@Override
			public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
				System.out.println("拒绝你"+r);
				
			}
		});

//特点:
		//1.没有核心线程
		//2.临时线程的数量是Integer.Max_value,由于单台服务器所能承载线程数量远远达不到这个值,所以认为这个线程池无限大
		//3.工作队列是一个同步队列
		//大池子小队列
		//使用场景:高并发短任务
		//实际场景中,在测试阶段会填充一个无用的空请求
		ExecutorService es=Executors.newCachedThreadPool();

//特点:
		//1.没有临时线程,都是核心线程
		//2.工作队列是阻塞式链式队列,没有指定容量,则意味着可以存储无限多的请求
		//小池子大队列
		//使用场景:并发少,长任务  云盘 下载队列
		ExecutorService es=Executors.newFixedThreadPool(5);
		//Callable只能通过线程池启动
		//将返回结果封装成了Future
		Future<String> f=es.submit(new CDemo());
		//将结果从Future中解析出来
		System.out.println(f.get());
		es.shutdown();
		
	}
}
//表示返回值类型
class CDemo implements Callable<String>{

	@Override
	public String call() throws Exception {
		
		return "Success";
	}
	
}

//这个方法不能起到定时效果
		//ses.submit(x);
		//推迟指定时间执行
		//ses.schedule(new ScheduledThread(), 5, TimeUnit.SECONDS);
		//每隔五秒执行一次
		//如果线程执行时间大于间隔时间,则以执行时间为准
		ses.scheduleAtFixedRate(new ScheduledThread(), 0, 5, TimeUnit.SECONDS);//从上一次执行开始计时
		//ses.scheduleWithFixedDelay(new ScheduledThread(), 0, 5, TimeUnit.SECONDS);//从上一次结束开始计时

public class ForkJoinPoolDemo {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
//		long sum=0;
//		
//		for(long i=1;i<10000000000000000L;i++){
//			sum+=i;
//			System.out.println(sum);
//		}
		ForkJoinPool pool=new ForkJoinPool();
		Future<Long> f =pool.submit(new Sum(1,100000000000L));
		System.out.println(f.get());
		pool.shutdown();
	}
}
class Sum extends RecursiveTask<Long>{
	private long start;
	private long end;
	public Sum(long start, long end) {
		super();
		this.start = start;
		this.end = end;
	}
	//进行分叉合并
	@Override
	protected Long compute() {
		if(end -start <=10000){
			long sum=0;
			for(long i=start; i<=end;i++){
				sum+=i;
			}
			return sum;
		}else{
			long mid=(start+end)/2;
			Sum left=new Sum(start,mid);
			Sum right=new Sum(mid+1,end);
			//分叉
			left.fork();
			right.fork();
			//合并
			 return left.join()+ right.join();
		}
		
	}
}

public class LockDemo {
	static int i=0;
	public static void main(String[] args) throws InterruptedException {
		Lock l=new ReentrantLock();
		new Thread(new Add(l)).start();
		new Thread(new Add(l)).start();
		Thread.sleep(3000);
		System.out.println(i);
	}
}

class Add implements Runnable{
	private Lock l;
	
	public Add(Lock l) {
		super();
		this.l = l;
	}

	@Override
	public void run() {
		//加锁
			l.lock();
			for(int i=0;i<10000;i++){
				LockDemo.i++;
			}	
			//解锁
			l.unlock();
		
		
	}
	
}

public static void main(String[] args) throws InterruptedException {
	CountDownLatch cdl=new CountDownLatch(5);
	new Thread(new Teacher(cdl)).start();
	new Thread(new Student(cdl)).start();
	new Thread(new Student(cdl)).start();
	new Thread(new Student(cdl)).start();
	new Thread(new Student(cdl)).start();
	//在计数归零前,需要让当前线程陷入等待
	cdl.await();
	//放在主函数中
	//在java中主函数所在的类默认就是一个单独的线程
	System.out.println("开始考试");

}
}
class Teacher implements Runnable{
	private CountDownLatch cd1;
	

	public Teacher(CountDownLatch cd1) {
		super();
		this.cd1 = cd1;
	}


	@Override
	public void run() {
		try {
			Thread.sleep((long) (Math.random()*10000));
			System.out.println("老师到");
			//计数减少一个
			cd1.countDown();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
}
class Student implements Runnable{
	private CountDownLatch cd1;
	public Student(CountDownLatch cd1) {
		super();
		this.cd1 = cd1;
	}
	@Override
	public void run() {
		try {
			Thread.sleep((long) (Math.random()*10000));
			System.out.println("考生到");
			//计数减少一个
			cd1.countDown();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
}

public class CyclicBarrierDemo {
	public static void main(String[] args) {
		CyclicBarrier cb=new CyclicBarrier(6);
		new Thread(new Runner(cb),"1号").start();
		new Thread(new Runner(cb),"2号").start();
		new Thread(new Runner(cb),"3号").start();
		new Thread(new Runner(cb),"4号").start();
		new Thread(new Runner(cb),"5号").start();
		new Thread(new Runner(cb),"6号").start();
	}
}
class Runner implements Runnable{
	private CyclicBarrier cb;
	public Runner(CyclicBarrier cb) {
		super();
		this.cb = cb;
	}
	@Override
	public void run() {
		try {
			Thread.sleep((long) (Math.random()*10000));
			String name=Thread.currentThread().getName();
			System.out.println(name+"远动员到了起跑线");
			//让当前线程进入阻塞,并且减少一
			cb.await();
			System.out.println(name+"运动员跑了出去");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}

public class ExchangerDemo {
	public static void main(String[] args) {
		Exchanger<String> ex=new Exchanger<>();
		new Thread(new Producer(ex)).start();
		new Thread(new Consumer(ex)).start();
	}
}
//生产者
class Producer implements Runnable{
	private Exchanger<String> ex;
	 
	public Producer(Exchanger<String> ex) {
		super();
		this.ex = ex;
	}

	@Override
	public void run() {
		String info="商品";
		//生产者需要将商品交换给消费者
		//同时,生产者应该收到消费者换过来的前钱
		//利用交换机来交换数据
		try {
			String msg=ex.exchange(info);
			System.out.println("生产者收到了消费者换过来的:"+msg);
			
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
}
//消费者
class Consumer implements Runnable{
	private Exchanger<String> ex;
	public Consumer(Exchanger<String> ex) {
		super();
		this.ex = ex;
	}

	@Override
	public void run() {
		String info="钱";
		//消费者需要将钱交还给生产者
		//同时,消费者应该收到生产者换过来的商品
		try {
			String msg=ex.exchange(info);
			System.out.println("消费者收到生产者交换来的"+msg);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}

public class SemaphoreDemo {
	public static void main(String[] args) {
		//表示有五张桌子
		Semaphore s=new Semaphore(5);
		for(int i=0;i<8;i++){
			new Thread(new Eater(s)).start();
		}
	}
}
class Eater implements Runnable{
	Semaphore s;
	
	public Eater(Semaphore s) {
		super();
		this.s = s;
	}

	@Override
	public void run() {
		
		try {
			//表示获取一个信号
			//如果信号减为0,则后续的线程就会被阻塞
			s.acquire();
			//当桌子被占用的时候,意味着可以使用的资源减少,对外暴露的信号减少一个
			System.out.println("过来一个人就餐,占用一张桌子");
			Thread.sleep((long) (Math.random()*10000));
			//当有桌子被空出的时候,就意味着对外暴露的信号多了一个
			//释放被占用信号
			s.release();
			System.out.println("一个人吃完了,空出一个桌子");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}

public class AtomicDemo {
	//static int i=0;
	//对外提供了很多线程安全的方法来操作该值
	//AtomicInteger在底层会对操作进行加锁
	//Jdk1.8开始,AtomicInteger采用Cas保证线程安全
	static AtomicInteger ai=new AtomicInteger(0);
	//原子性引用类型AtomicReference<V>
	public static void main(String[] args) throws InterruptedException {
		CountDownLatch cd1=new CountDownLatch(3);
		new Thread(new Add(cd1)).start();
		new Thread(new Add(cd1)).start();
		new Thread(new Add(cd1)).start();
		cd1.await();
		System.out.println(ai);
	}
}
class Add implements Runnable{
CountDownLatch cd1;

	public Add(CountDownLatch cd1) {
	super();
	this.cd1 = cd1;
}

	@Override
	public void run() {
		for(int i=0;i<10000;i++){
			//AtomicDemo.i++;
			AtomicDemo.ai.incrementAndGet();//++i
			
		}
		cd1.countDown();
	}
}