happyfish

happyfish 查看完整档案

北京编辑西北大学  |  计算机科学与技术 编辑西安川沃科技  |  java工程师 编辑填写个人主网站
编辑

个人动态

happyfish 分享了头条 · 2017-08-02

对String.intern做一些基础补充

赞 0 收藏 1 评论 0

happyfish 分享了头条 · 2017-07-15

写的不错。

赞 0 收藏 1 评论 0

happyfish 发布了文章 · 2017-07-12

深入剖析ThreadPool的运行原理

线程在执行任务时,正常的情况是这样的:

Thread  t=new Thread(new  Runnable() {            
            @Override
            public void run() {
                // TODO Auto-generated method stub    
            }
        });
        
        t.start();
        

  Thread 在初始化的时候传入一个Runnable,以后就没有机会再传入一个Runable了。那么,woker作为一个已经启动的线程。是如何不断获取Runnable的呢?
这个时候可以使用一个包装器,将线程包装起来,在Run方法内部获取任务。

public final class Worker implements Runnable {
    Thread thread = null;
    Runnable task;
    private BlockingQueue<Runnable> queues;
    public Worker(Runnable task, BlockingQueue<Runnable> queues) {
        this.thread = new Thread(this);
        this.task = task;
        this.queues = queues;
    }
    public void run() {
        if (task != null) {
            task.run();
        } 
            try {
                while (true) {
                    task = queues.take();
                    if (task != null) {
                        task.run();
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    public void start() {
        this.thread.start();
    }
}

public class Main {
    public static void main(String[] args) {
        BlockingQueue<Runnable> queues=new ArrayBlockingQueue<Runnable>(100);
        Worker  worker=new Worker(new Runnable() {
            public void run() {
                System.out.println("hello!!! ");
                try {
                    Thread.currentThread().sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }            
            }
        }, queues);
        worker.start();
        for(int i=0;i<100;i++){
            queues.offer(new Runnable() {
                public void run() {
                    System.out.println("hello!!! ");
                    try {
                        Thread.currentThread().sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

}

  这样我们就简单地实现了一个“线程池”(可以将这个“线程池”改造成官方的模式,不过可以自己尝试一下)。ThreadPool的这种实现模式是并发编程中经典的Cyclic Work Distribution模式。
  那么,这种实现的线程池性能如何呢?
  由于其任务队列使用的是阻塞队列,在队列内部是自旋的。Reeteenlok是改进的CLH队列。自旋锁会耗费一定CPU的资源,在拥有大量任务执行下的情况下比较有效。而且,线程池中的线程并没有睡眠,而是进入了自旋状态。

CPU的线程与关系

  如果是不支持超线程的CPU,在同一时刻的确只能处理2个线程,但是并不意味着双核的CPU只能处理两个线程,它可以通过切换上下文来执行多个线程。比如我只有一个大脑,但是我要处理5个人提交的任务,我可以处理完A的事情后,把事情的中间结果保存下,然后再处理B的,然后再读取A的中间结果,处理A的事情。

JDK中的线程池实现分析

  Woker自身继承了Runnable,并对Thread做了一个包装。Woker代码如下所示:

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {

        private static final long serialVersionUID = 6138294804551838833L;

    
        Runnable firstTask;
   
        volatile long completedTasks;

 
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
        public void run() {
            runWorker(this);
        }
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }

        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void lock()        { acquire(1); }
        public boolean tryLock()  { return tryAcquire(1); }
        public void unlock()      { release(1); }
        public boolean isLocked() { return isHeldExclusively(); }

        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

  execute(Runnable command)方法内部是这样的:

public void execute(Runnable command) {
       if (command == null)
           throw new NullPointerException();
     
       int c = ctl.get();
       if (workerCountOf(c) < corePoolSize) {
           if (addWorker(command, true))
               return;
           c = ctl.get();
       }
       if (isRunning(c) && workQueue.offer(command)) {
           int recheck = ctl.get();
           if (! isRunning(recheck) && remove(command))
               reject(command);
           else if (workerCountOf(recheck) == 0)
               addWorker(null, false);
       }
       else if (!addWorker(command, false))
           reject(command);
   }

  ctl一个合并类型的值。将当前线程数和线程池状态通过数学运算合并到了一个值。具体是如何合并的可以参看一下源码,这里就不叙述了。继续向下走:

if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }

  可以看到,如果当前线程数量小于了核心线程数量corePoolSize,就直接增加线程处理任务。与队列没有关系。但是紧接着又检查了一遍状态,因为在这个过程中,别的线程也可能在添加任务。继续向下走:


  if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        
        

  可以看到如果线程池是运行态的,就把线程添加到任务队列。workQueue是构造函数传递过来的,可以是有界队列,也可以是无界队列。可以看出来,队列如果是无界的,直接往队列里面添加任务,这个时候,线程池中的线程也不会增加,一直会等于核心线程数。
  如果队列是有界的,就尝试直接新增线程处理任务,如果添加任务失败,就调用reject方法来处理添加失败的任务:

 else if (!addWorker(command, false))
            reject(command);

  来看看addWorker是如何实现的,逻辑流程已经直接在注释中说明了。

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            //如果状态大于SHUTDOWN,不再接受新的任务,直接返回
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

/**根据core来判断,如果当前线程数量大于corePoolSize或者最大线程数,直接返回。添加任务失败。
**如果队列是有界的或者任务添加到队列失败(参数core是false),那么就会新开一个线程处理业务,但如果线程已经大于了maximumPoolSize,就会出现添加失败,返回false。
*/

            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;

                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

  如果创建失败的情况下会调用addWorkerFailed方法,从而将减少实际线程数。

addWorker中for循环的意义

  在addWorker中有这么一段代码,表示为当前线程数加1:

  private boolean compareAndIncrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }

  由于多线程可能同时操作。expect值可能会变化。仅仅一次的操作compareAndIncrementWorkerCount可能一次并不会成功,而且,一个线程在执行addWork的过程中间,另外一个线程假设直接shotdown这个线程池。for循环的存在可以保证状态一定是一致的。

任务的执行

在Worker中间实际上是调用的runWorker方法来执行的具体业务:

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

  可以看到while循环不断的从队列中取出任务执行。如果task==null 并且getTask()等于null的话,那么就会跳出循环,进入到processWorkerExit,run方法执行完毕以后,这个线程也被销毁了。但是为什么在各自的线程执行,为什么还需要加锁呢?答案是因为要线程池需要判断这个线程是否在执行任务。在interruptIdleWorkers方法中,要中断那写目前空闲的线程,通过当前Worker是否获得了锁就能判断这个worker是否是空闲的:

 private void interruptIdleWorkers(boolean onlyOne) {
      final ReentrantLock mainLock = this.mainLock;
      mainLock.lock();
      try {
          for (Worker w : workers) {
              Thread t = w.thread;
              if (!t.isInterrupted() && w.tryLock()) {
                  try {
                      t.interrupt();
                  } catch (SecurityException ignore) {
                  } finally {
                      w.unlock(); //中断不起作用。interrupt()对于自旋锁是不起作用的。只是逻辑上被阻塞,
                  }
              }
              if (onlyOne)
                  break;
          }
      } finally {
          mainLock.unlock();
      }
  }

  可以看到,如果w.tryLock()可以获取到锁,那么就意味着当前的 Woker并没有处理任务(没有进入到循环里面或者被getTask方法所阻塞,无法获取锁)。
Work之所以继承AbstractQueuedSynchronizer,而不去使用ReentrantLock。是因为ReentrantLock是可重入锁,在调用lock方法获取锁之后,再调用tryLock()还是会返回true。

public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        lock.lock();
        System.out.println(lock.tryLock());        
    }

输出结果是true,所以使用ReentrantLock则难以判断当前Worker是否在执行任务。

线程超时allowCoreThreadTimeOut、keepAliveTime以及线程死亡

  在上面的interruptIdleWorkers方法中,线程被中断。普通的线程被中断会导致线程继续执行,从而run方法运行完毕,线程退出。

对于一个没有被阻塞的线程,中断是不起作用的。中断在如下线程被阻塞的方法中起作用:
the wait(),
wait(long),
wait(long, int)
join(),
join(long),
join(long, int),
sleep(long),
or sleep(long, int)
LockSupport.park(Object object);
LockSupport.park();

  ,如果唤醒这些被阻塞的线程,从而能使得run方法继续执行,当run方法执行完毕,那么线程也就终结死亡。但是对于ReentrantLock和AbstractQueuedSynchronizer这种自旋+CAS实现的“逻辑锁”,是不起作用的。
而且runWork本身也是While循环,靠中断是无法退出循环的。

  但是在ThreadPoolExecutor的构造函数中,有一个允许设置线程超时allowCoreThreadTimeOut参数的方法。如果允许超时,多于corePoolSize的线程将会在处在空闲状态之后存活keepAliveTime时长后终止。因此有了一个allowCoreThreadTimeOut方法:

 public void allowCoreThreadTimeOut(boolean value) {
        if (value && keepAliveTime <= 0)
            throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
        if (value != allowCoreThreadTimeOut) {
            allowCoreThreadTimeOut = value;
            if (value)
                interruptIdleWorkers();
        }
    }

   正如上面提到的一样,允许allowCoreThreadTimeOut并且调用interruptIdleWorkers方法并不能使线程退出。那么线程池又如何杀掉这个线程呢?
  没错,就是getTask方法。只有当getTask返回null的时候才能跳出While循环,run方法运行完毕,那么线程自然而然就死亡了。getTask方法如下所示:

private Runnable getTask() {
      boolean timedOut = false; // Did the last poll() time out?
      for (;;) {
          int c = ctl.get();
          int rs = runStateOf(c);

          // Check if queue empty only if necessary.
          if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
              decrementWorkerCount();
              return null;
          }

          int wc = workerCountOf(c);
          // Are workers subject to culling?
          boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
          if ((wc > maximumPoolSize || (timed && timedOut))
              && (wc > 1 || workQueue.isEmpty())) {
              if (compareAndDecrementWorkerCount(c))
                  return null;
              continue;
          }
          try {
              Runnable r = timed ?
                  workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                  workQueue.take();
              if (r != null)
                  return r;
              timedOut = true;
          } catch (InterruptedException retry) {
              timedOut = false;
          }
      }
  }

  可以看到,如果线程池状态大于SHUTDOWN并且队列空,返回null,从而结束循环。(线程死亡)

  或者状态大于SHUTDOWN并且线程大于STOP(STOP一定大于SHUTDOWN,所以可以直接说线程大于STOP)返回null,从而结束循环。(线程死亡)
再往下可以看到如果超过了maximumPoolSize,返回null,从而结束循环。(线程死亡)
超过keepAliveTime时间,任务对列没有数据而返回null。从而结束循环。(线程死亡)
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;保证了线程池至少留下corePoolSize个线程。

预定义的拒接处理协议

  在execute方法中,如果线程池拒绝添加任务,就会有一个钩子方法来处理被拒绝的任务。
可以自己定义,也可以使用线城池中默认的拒接处理协议。

AbortPolicy :直接抛出RejectedExecutionException异常;

CallerRunsPolicy:谁调用的execute方法,谁就执行这个任务;

DiscardPolicy:直接丢弃,什么也不做;

DiscardOldestPolicy:丢弃对列中间最老的任务,执行新任务。

有什么问题或者建议,可以加入小密圈和我一起讨论,或者在简书留言,欢迎喜欢和打赏。

最后向大家安利一本我写的关于Java并发的书籍:Java并发编程系统与模型,个人觉得写得不错,比较通俗易懂,非常适合初学者,百度阅读可以下载电子书。

查看原文

赞 1 收藏 6 评论 2

happyfish 分享了头条 · 2017-07-12

一本系统介绍Java并发编程方面的书籍。囊括了从最基本的并发原理到顶层API的使用,以及到最后的深入实践。目前整个市面上的有关java并发方面的书籍要么就是以点概面,要么就是没有从实际生产中出发,不能学以致用。本书不但尝试从How to Use去介绍并发,更从知识是如...

赞 0 收藏 0 评论 0

happyfish 分享了头条 · 2017-07-12

分析的比较细致,原理讲的比较透彻

赞 0 收藏 2 评论 0

happyfish 提出了问题 · 2017-06-29

Spring 定时任务如何实现每周一某个时间执行?

比如通过@Scheduled(fixedRate = 900000)注解就可以完成那种的。

关注 6 回答 6

happyfish 回答了问题 · 2017-05-15

有zookeeper相关的大神么?使用ZKClient创建节点,但是服务器中并没有显示?

自问自答吧。因为我创建的是临时节点,临时节点在绘画断开以后会自动删除,估计这个 ls / 命令应该是列举的持久节点。

关注 1 回答 1

happyfish 提出了问题 · 2017-05-15

有zookeeper相关的大神么?使用ZKClient创建节点,但是服务器中并没有显示?

我按照zk权威指南上的代码创建节点:

package com.zkstudy;

import java.io.IOException;
import java.util.Random;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class Master implements Watcher {
    private ZooKeeper zk = null;
    private String host;
    private String serverId = null;
    private boolean isLeader = false;

    public Master(String host) {
        this.host = host;
        Random radom = new Random();
        serverId = Integer.toHexString(radom.nextInt());
    }

    public void start() {

        try {
            this.zk = new ZooKeeper(host, 15000, this);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void createMaster() {

        while (true) {

            try {
                zk.create("/master", serverId.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                isLeader = true;
                break;
            } catch (KeeperException e) {
                e.printStackTrace();
                isLeader = false;
                break;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (checkMaster()) {
                break;
            }

        }

    }

    public boolean checkMaster() {

        while (true) {
            Stat stat = new Stat();
            try {
                byte[] data = zk.getData("/master", false, stat);
                isLeader = new String(data).equals(serverId);
                return true;
            } catch (KeeperException e) {
                e.printStackTrace();
                return false;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void process(WatchedEvent event) {
        System.out.println("watcher thread:---------->" + Thread.currentThread().getId());
        System.out.println("event:---------->" + event);
    }

    public static void main(String[] args) throws InterruptedException {
        Master master = new Master("123.206.28.37:2181");
        master.start();
        master.createMaster();
        if (master.isLeader) {
            System.out.println("i am  master");
        }
        System.out.println("main thread:---------->" + Thread.currentThread().getId());
        Thread.sleep(5000);
    }
}

然后我运行了两遍。出现了如下问题:

watcher thread:---------->10
event:---------->WatchedEvent state:SyncConnected type:None path:null
org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode = NodeExists for /master
main thread:---------->1
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:119)
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:51)
    at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:783)
    at com.zkstudy.Master.createMaster(Master.java:40)
    at com.zkstudy.Master.main(Master.java:83)

提示master节点已经存在。
但是我在命令行里面查看却没有发现master节点:

[zk: localhost:2181(CONNECTED) 10] ls
[zk: localhost:2181(CONNECTED) 11] ls  /
[zk, zookeeper]
[zk: localhost:2181(CONNECTED) 12] 

请问这是为什么呢?zk使用的是单节点,不是伪集群。

关注 1 回答 1

happyfish 提出了问题 · 2017-04-14

Netty的future.channel().closeFuture().sync();到底有什么用?

我看到很多Netty的例子都在末尾加上了这句话:future.channel().closeFuture().sync();

比如:

public class TimeServer {
    private int count = 0;

    public void bind(int port) {
        try {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workGroup = new NioEventLoopGroup();
            ServerBootstrap b = new ServerBootstrap(); // (2)
            b.group(bossGroup, workGroup).channel(NioServerSocketChannel.class) // (3)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel arg0) throws Exception {
                            arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));
                            arg0.pipeline().addLast(new StringDecoder());
                            arg0.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                    // ByteBuf buf = (ByteBuf) msg;
                                    // byte[] req = new
                                    // byte[buf.readableBytes()];
                                    // buf.readBytes(req);
                                    // String body = new String(req, "UTF-8");
                                    System.out.println(
                                            "The Time Server  Received order:" + msg + "; the  counter is:" + ++count);

                                    // String currentTime = "QUERY TIME
                                    // ORDER".equalsIgnoreCase(body)
                                    // ? new
                                    // Date(System.currentTimeMillis()).toString()
                                    // : "BAD ORDER";
                                    //
                                    // currentTime = currentTime +
                                    // System.getProperty("line.separator");
                                    // ByteBuf resp =
                                    // Unpooled.copiedBuffer(currentTime.getBytes());
                                    // ctx.writeAndFlush(resp);
                                }
                            });
                        }
                    });
            ChannelFuture future = b.bind(port).sync();

            System.out.println("Server start listen at " + port);
            future.channel().closeFuture().sync();
            
            System.out.println("执行到这里 " + port);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new TimeServer().bind(10000);

    }
}

但是我看这行代码一直没有执行。请问这是怎么回事呢?

关注 2 回答 2

happyfish 赞了回答 · 2017-04-14

解决使用Netty Demo报错

在netty4中,对象的生命周期由引用计数器控制,ByteBuf就是如此,每个对象的初始化引用计数为1,调用一次release方法,引用计数器会减1,当尝试访问计数器为0的,对象时,会抛出IllegalReferenceCountException,正如ensureAccessible的实现,更加详细的解释可以参考官方文档

AbstractByteBuf.java

protected final void ensureAccessible() {
        if (refCnt() == 0) {
            throw new IllegalReferenceCountException(0);
        }
    }

注意TZ的TimeHandler类中的 super.channelRead(ctx, msg);这行代码。追踪调用路径,

private void invokeChannelRead(Object msg) {
        try {
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    }

最终调用的代码是:ReferenceCountUtil.release(msg)

public static boolean release(Object msg) {
        if (msg instanceof ReferenceCounted) {
            return ((ReferenceCounted) msg).release();
        }
        return false;
    }

也就是每次super.channelRead(ctx, msg);后,ByteBuf就会调用release()方法,计数器减一,然后在 buf.readBytes(butfs);这行代码就会校验ensureAccessible(),计数器为0,netty认为ByteBuf对象已经释放,就抛出异常。

解决方案:

去掉TimeHandler中这行代码 super.channelRead(ctx, msg);
ByteBuf对象谁处理谁释放。

关注 3 回答 2

认证与成就

  • 获得 109 次点赞
  • 获得 375 枚徽章 获得 21 枚金徽章, 获得 154 枚银徽章, 获得 200 枚铜徽章

擅长技能
编辑

开源项目 & 著作
编辑

(゚∀゚ )
暂时没有

注册于 2014-04-16
个人主页被 1k 人浏览