javase(11)_juc并发库

一、传统线程技术

public static void main(String[] args) {
    Thread thread = new Thread(){
        @Override
        public void run() {
            while(true){
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("1:" + Thread.currentThread().getName());
            }
        }
    }.start();
    
    Thread thread2 = new Thread(new Runnable(){
        @Override
        public void run() {
            while(true){
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("1:" + Thread.currentThread().getName());
            }                
        }
    }).start();
}

二、传统定时器技术

一种工具,线程用其安排以后在后台线程中执行任务,可安排任务执行一次,或者定期重复执行,与每个 Timer 对象相对应的是单个后台线程,用于顺序地执行所有计时器任务.

public static void main(String[] args) throws InterruptedException {//下面两个会同时执行,Timer也是一个线程
        new Timer().schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("bombing!");
            }
        }, 10000,1000);      //10s之后开始执行任务,之后每一秒执行一次
        new Thread(){
            @Override
            public void run() {
                while(true){
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {}
                    System.out.println("wangwei");
                }
            }
        }.start();
}

三、ThreadLocal类

在同步机制中,通过对象的锁机制保证同一时间只有一个线程访问变量.这时该变量是多个线程共享的,使用同步机制要求程序慎密地分析什么时候对变量进行读写,什么时候需要锁定某个对象,什么时候释放对象锁等繁杂的问题,程序设计和编写难度相对较大.而ThreadLocal则从另一个角度来解决多线程的并发访问.ThreadLocal会为每一个线程提供一个独立的变量副本,从而隔离了多个线程对数据的访问冲突.因为每一个线程都拥有自己的变量副本,从而也就没有必要对该变量进行同步了.ThreadLocal提供了线程安全的共享对象,在编写多线程代码时,可以把不安全的变量封装进ThreadLocal.

当然ThreadLocal并不能替代同步机制,两者面向的问题领域不同.同步机制是为了同步多个线程对相同资源的并发访问,是为了多个线程之间进行通信 的有效方式;而ThreadLocal是隔离多个线程的数据共享,从根本上就不在多个线程之间共享资源(变量),这样当然不需要对多个线程进行同步了.所以,如果你需要进行多个线程之间进行通信,则使用同步机制;如果需要隔离多个线程之间的共享冲突,可以使用ThreadLocal,这将极大地简化你的程序,使程序更加易读、简洁.

ThreadLocal底层其实就是把当前线程和要往ThreadLocal中存的值装在一个Map中.

ThreadLocal用于session

private static final ThreadLocal<Session> threadSession = new ThreadLocal<Session>();
 
public static Session getSession() throws InfrastructureException {
    Session s = (Session) threadSession.get();
    try {
        if (s == null) {
            s = getSessionFactory().openSession();
            threadSession.set(s);
        }
    } catch (HibernateException ex) {
        throw new InfrastructureException(ex);
    }
    return s;
}

ps:参考jdbc节中ThreadLocal用于service层,controler层,甚至一个会话中的事务管理.

四、接口 Executor

执行已提交的 Runnable对象的任务.此接口提供一种将任务提交与每个任务将如何运行的机制(包括线程使用的细节、调度等)分离开来的方法.通常使用Executor而不是显式地创建线程.例如下所示,而不是为一组任务中的每个任务调用 new Thread(new(RunnableTask())).start().

import java.util.concurrent.Executor;

class TestRunnable implements Runnable{
    @Override
    public void run() {
        for(int i=0;i<10;i++)
            System.out.println("ww");
    }
}
class MyExecutor implements Executor{
    @Override
    public void execute(Runnable command) {
        new Thread(command).start();
    }
}
public class TestExecutor extends Thread{
    public static void main(String[] args) {
        TestRunnable t = new TestRunnable();
        Executor executor = new MyExecutor();
        executor.execute(t);
    }
}

五、线程池使用

大多数服务器应用程序(如web服务器、POP服务器、数据库服务器或文件服务器)代表远程客户机处理请求,这些客户机通过(http协议,ftp协议、pop协议)或者JMS队列等,不管以什么方式连接到服务器.对于每个请求,通常要进行少量处理(获得该文件的代码块,并将其发送回 socket),但是可能会有大量(且不受限制)的客户机请求服务.用于构建服务器应用程序的简单化模型会为每个请求创建新的线程.下列代码段实现简单的Web服务器,它接受端口80的socket 连接,并创建新的线程来处理请求.不幸的是,该代码不是实现Web服务器的好方法,因为在重负载条件下它将失败,停止整台服务器.

class UnreliableWebServer {
    public static void main(String[] args) throws IOException {
        ServerSocket socket = new ServerSocket(80);
        while (true) {
            final Socket connection = socket.accept();
            Runnable r = new Runnable() {
                public void run() {
                    handleRequest(connection);
                }
            };
            new Thread(r).start();
        }
    }
}

无论如何,这样使用资源可能会损害性能.如果创建过多线程,其中每个线程都将占用一些 CPU 时间,结果将使用许多内存来支持大量线程,每个线程都运行得很慢.这样就无法很好地使用计算资源,管理一大组小任务的标准机制是组合工作队列和线程池.工作队列就是要处理的任务的队列,前面描述的 Queue 类完全适合.线程池是线程的集合,每个线程都提取公用工作队列.当一个工作线程完成任务处理后,它会返回队列,查看是否有其他任务需要处理.如果有,它会转移到下一个任务,并开始处理,如下是一个使用线程池的简单网络服务:

class NetworkService implements Runnable {
    private final ServerSocket serverSocket;
    private final ExecutorService pool;

    public NetworkService(int port, int poolSize)
        throws IOException {
      serverSocket = new ServerSocket(port);
      pool = Executors.newFixedThreadPool(poolSize);
    }
 
    public void run() { // run the service
      try {
        for (;;) {
          pool.execute(new Handler(serverSocket.accept()));
        }
      } catch (IOException ex) {
        pool.shutdown();
      }
    }
  }

  class Handler implements Runnable {
    private final Socket socket;
    Handler(Socket socket) { this.socket = socket; }
    public void run() {
      // read and service request on socket
    }
 }

几种类型的线程池

public class ThreadPoolTest {
    public static void main(String[] args) {
        //可重用固定线程数的线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(3);   
        //可根据需要创建新线程的缓存线程池
        //ExecutorService threadPool = Executors.newCachedThreadPool(); 
        //单个线程的线程池
        //ExecutorService threadPool = Executors.newSingleThreadExecutor(); 
        //一个可以在给定延迟后运行命令或者定期地执行的线程池
        ScheduledExecutorService  ScheduledThreadPool = Executors.newScheduledThreadPool(3);
        for(int i=1;i<=10;i++){
            final int task = i;
            threadPool.execute(new Runnable(){
                @Override
                public void run() {
                    for(int j=1;j<=10;j++){
                        System.out.println(Thread.currentThread().getName() + " is looping of " + j 
　　　　　　　　　　　　　　+ " for  task of " + task);
                    }
                }
            });
        }
        
        ScheduledThreadPool.scheduleAtFixedRate(new Runnable(){
            public void run(){
                System.out.println("bombing!");
            }
        }, 10, 1, TimeUnit.SECONDS);
        
    }
}

六、互斥锁

public class LockTest {
    public static void main(String[] args) {
        new LockTest().init();
    }
    private void init(){
        final Outputer outputer = new Outputer();
        new Thread(new Runnable(){
            @Override
            public void run() {
                while(true){
                    outputer.output("zhangxiaoxiang");
                }
            }
        }).start();
        
        new Thread(new Runnable(){
            @Override
            public void run() {
                while(true){
                    outputer.output("lihuoming");
                }
            }
        }).start();
    }  
}
class Outputer{
    Lock lock = new ReentrantLock();
    public void output(String name){
        int len = name.length();
        lock.lock();
        try{
            for(int i=0;i<len;i++){
                System.out.print(name.charAt(i));
            }
            System.out.println();
        }finally{
            lock.unlock();
        }
    }
}

七、读写锁

与互斥锁相比,读-写锁允许对共享数据进行更高级别的并发访问.虽然一次只有一个线程(writer 线程)可以修改共享数据,但在许多情况下,任何数量的线程可以同时读取共享数据(reader 线程),读-写锁利用了这一点.使用读-写锁所允许的并发性增强将带来更大的性能提高.

public class ReadWriteLockTest {
    public static void main(String[] args) {
        final Queue q = new Queue();
        for(int i=0;i<3;i++){
            new Thread(){
                public void run(){
                    while(true){
                        q.get();                        
                    }
                }
            }.start();

            new Thread(){
                public void run(){
                    while(true){
                        q.put(new Random().nextInt(10000));
                    }
                }            
            }.start();
        }
    }
}

class Queue{
    private Object data = null;//共享数据,只能有一个线程写该数据,但可以多个线程同时读该数据.
    ReadWriteLock rwl = new ReentrantReadWriteLock();
    public void get(){
        rwl.readLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"read data!");
            Thread.sleep((long)(Math.random()*100));
            System.out.println(Thread.currentThread().getName()+"read:"+data);            
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally{
            rwl.readLock().unlock();
        }
    }
    
    public void put(Object data){
        rwl.writeLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"write data!");                    
            Thread.sleep((long)(Math.random()*1000));
            this.data = data;        
            System.out.println(Thread.currentThread().getName()+"havewrite"+data);                    
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally{
            rwl.writeLock().unlock();
        }
    }
}

八、Condition

Condition将Object监视器方法wait,notify,notifyll分解成截然不同的对象,以便通过将这些对象与任意lock实现组合使用,改写生产者消费者模式:

class Product {
    private int id;
    Product(int id) {
        this.id = id;
    }
    public String toString() {
        return "Product [id=" + id + "]";
    }
}

class Box {
    Lock lock = new ReentrantLock();
    Condition putConditon = lock.newCondition();
    Condition popConditon = lock.newCondition();
    Product[] p = new Product[10]; // 此处可以定义为一个队列
    int index = 0;

    public void put(Product pro) {
        try {
            lock.lock();
            while (index == p.length) {
                try {
                    putConditon.await();
                } catch (InterruptedException e) {
                }
            }
            p[index] = pro;
            index++;
            popConditon.signal();
        } finally {
            lock.unlock();
        }
    }

    public Product pop() {
        try {
            lock.lock();
            while (index == 0) {
                try {
                    popConditon.await();
                } catch (InterruptedException e) {
                }
            }
            putConditon.signal();
            index--;
            return p[index];
        } finally {
            lock.unlock();
        }
    }
}

class Producter implements Runnable {
    Box box = null;
    Producter(Box box) {
        this.box = box;
    }
    public void run() {
        for (int i = 0; i < 20; i++) { // 每个生产者生产20个
            Product pro = new Product(i);
            box.put(pro);
            System.out.println(Thread.currentThread().getName()+"生产"+pro);
        }
    }
}

class Customer implements Runnable {
    Box box = null;
    Customer(Box box) {
        this.box = box;
    }
    public void run() {
        while (true) {
            Product pro = box.pop();
            System.out.println(Thread.currentThread().getName()+"消费"+pro);
        }
    }
}

public class TestCondions {
    public static void main(String[] args) {
        Box box = new Box();
        Producter p = new Producter(box);
        Customer c = new Customer(box);
        new Thread(p).start();
        new Thread(p).start();
        new Thread(p).start();
        new Thread(c).start();
        new Thread(c).start();
        new Thread(c).start();
    }
}

九、Semaphore

通常用于限制可以访问某些资源(物理或逻辑的)的线程数目,可以和线程池组合使用.

public class SemaphoreTest {
    public static void main(String[] args) {
        ExecutorService service = Executors.newCachedThreadPool();
        final  Semaphore sp = new Semaphore(3);
        for(int i=0;i<10;i++){
            Runnable runnable = new Runnable(){
                    public void run(){
                    try {
                        sp.acquire();
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                    try {
                        //处理业务
                        Thread.sleep((long)(Math.random()*10000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    sp.release();
                }
            };
            service.execute(runnable);            
        }
    }
}

十、CyclicBarrier、Exchanger、CountDownLatch、Callable与Future的应用,略.

ps:客户连接请求队列

在客户/服务器通信模式中,服务器端需要创建监听特定端口的ServerSocket,ServerSocket负责接收客户连接请求.本章首先介绍ServerSocket类的各个构造方法,以及成员方法的用法,接着介绍服务器如何用多线程来处理与多个客户的通信任务.

本章提供线程池的一种实现方式.线程池包括一个工作队列和若干工作线程.服务器程序向工作队列中加入与客户通信的任务,工作线程不断从工作队列中取出任务并执行它.

1、构造ServerSocket

ServerSocket的构造方法有以下几种重载形式:

◆ServerSocket()throws IOException
◆ServerSocket(int port) throws IOException
◆ServerSocket(int port, int backlog) throws IOException
◆ServerSocket(int port, int backlog, InetAddress bindAddr) throws IOException   

在以上构造方法中,参数port指定服务器要绑定的端口(服务器要监听的端口),参数backlog(积压)指定客户连接请求队列的长度,参数bindAddr指定服务器要绑定的IP地址.

2、绑定端口

除了第一个不带参数的构造方法以外,其他构造方法都会使服务器与特定端口绑定,该端口由参数port指定.例如,以下代码创建了一个与80端口绑定的服务器:

ServerSocket serverSocket=new ServerSocket(80);

3、设定客户连接请求队列的长度

当服务器进程运行时,可能会同时监听到多个客户的连接请求.例如,每当一个客户进程执行以下代码:

Socket socket=new Socket(www.javathinker.org,80);

就意味着在远程www.javathiker.org主机的80端口上,监听到了一个客户的连接请求.管理客户连接请求的任务是由操作系统来完成的.操作系统把这些连接请求存储在一个先进先出的队列中.许多操 作系统限定了队列的最大长度,一般为50.当队列中的连接请求达到了队列的最大容量时,服务器进程所在的主机会拒绝新的连接请求.只有当服务器进程通过 ServerSocket的accept()方法从队列中取出连接请求,使队列腾出空位时,队列才能继续加入新的连接请求.

对于客户进程,如果它发出的连接请求被加入到服务器的队列中,就意味着客户与服务器的连接建立成功,客户进程从Socket构造方法中正常返回.如果客户进程发出的连接请求被服务器拒绝,Socket构造方法就会抛出ConnectionException.

ServerSocket构造方法的backlog参数用来显式设置连接请求队列的长度,它将覆盖操作系统限定的队列的最大长度.值得注意的是,在以下几种情况中,仍然会采用操作系统限定的队列的最大长度:

◆backlog参数的值大于操作系统限定的队列的最大长度;
◆backlog参数的值小于或等于0;
◆在ServerSocket构造方法中没有设置backlog参数.

http://www.51cto.com/specbook/11/40196.htm