Wetts's blog

-agentlib:jdwp=transport=dt_socket,address=57333,suspend=n,server=y

public void extract() {
    logger.debug("开始" + getExtractorName() + "。。");
    //抽取邮件
    List<Article> articles = extractEmail();
    //添加文章
    for (Article article : articles) {
        addArticleOrComment(article);
    }

    //清空邮件
    cleanEmail();
    logger.debug("完成" + getExtractorName() + "。。");
}

public class QuickEmailToWikiExtractor extends AbstractExtractor {

    private ThreadPoolExecutor      threadsPool;
    private ArticleBlockingQueue<ExchangeEmailShallowDTO> emailQueue;

    public QuickEmailToWikiExtractor() {
        emailQueue= new ArticleBlockingQueue<ExchangeEmailShallowDTO>();
        int corePoolSize = Runtime.getRuntime().availableProcessors() * 2;
        threadsPool = new ThreadPoolExecutor(corePoolSize, corePoolSize, 10l, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(2000));
    }

    public void extract() {
        logger.debug("开始" + getExtractorName() + "。。");
        long start = System.currentTimeMillis();

        //抽取所有邮件放到队列里
        new ExtractEmailTask().start();

        // 把队列里的文章插入到Wiki
        insertToWiki();

        long end = System.currentTimeMillis();
        double cost = (end - start) / 1000;
        logger.debug("完成" + getExtractorName() + ",花费时间：" + cost + "秒");
    }

    /**
     * 把队列里的文章插入到Wiki
     */
    private void insertToWiki() {
        //登录wiki,每间隔一段时间需要登录一次
        confluenceService.login(RuleFactory.USER_NAME, RuleFactory.PASSWORD);

        while (true) {
            //2秒内取不到就退出
            ExchangeEmailShallowDTO email = emailQueue.poll(2, TimeUnit.SECONDS);
            if (email == null) {
                break;
            }
            threadsPool.submit(new insertToWikiTask(email));
        }
    }

     protected List<Article> extractEmail() {
        List<ExchangeEmailShallowDTO> allEmails = getEmailService().queryAllEmails();
        if (allEmails == null) {
            return null;
        }

        for (ExchangeEmailShallowDTO exchangeEmailShallowDTO : allEmails) {
            emailQueue.offer(exchangeEmailShallowDTO);
        }
        return null;
    }

    /**
     * 抽取邮件任务
     *
     * @author tengfei.fangtf
     */
    public class ExtractEmailTask extends Thread {
        public void run() {
            extractEmail();
        }
    }
}

/**
 * 总消息队列管理
 *
 * @author tengfei.fangtf
 */
public class MsgQueueManager implements IMsgQueue{
    private static final Logger LOGGER = LoggerFactory.getLogger(MsgQueueManager.class);

    /**
     * 消息总队列
     */
    public final BlockingQueue<Message> messageQueue;

    private MsgQueueManager() {
        messageQueue = new LinkedTransferQueue<Message>();
    }

    public void put(Message msg) {
        try {
            messageQueue.put(msg);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public Message take() {
        try {
            return messageQueue.take();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return null;
    }

}

/**
 * 分发消息，负责把消息从大队列塞到小队列里
 *
 * @author tengfei.fangtf
 */
static class DispatchMessageTask implements Runnable {
    @Override
    public void run() {
        BlockingQueue<Message> subQueue;
        for (;;) {
            //如果没有数据，则阻塞在这里
            Message msg = MsgQueueFactory.getMessageQueue().take();
            //如果为空，则表示没有Session机器连接上来，需要等待，直到有Session机器连接上来
            while ((subQueue = getInstance().getSubQueue()) == null) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            //把消息放到小队列里
            try {
                subQueue.put(msg);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

/**
 * 均衡获取一个子队列。
 *
 * @return
 */
public BlockingQueue<Message> getSubQueue() {
    int errorCount = 0;
    for (;;) {
        if (subMsgQueues.isEmpty()) {
            return null;
        }
        int index = (int) (System.nanoTime() % subMsgQueues.size());
        try {
            return subMsgQueues.get(index);
        } catch (Exception e) {
            //出现错误表示，在获取队列大小之后，队列进行了一次删除操作
            LOGGER.error("获取子队列出现错误", e);
            if ((++errorCount) < 3) {
                continue;
            }
        }
    }
}

//往消息队列里添加一条消息
IMsgQueue messageQueue = MsgQueueFactory.getMessageQueue();
Packet msg = Packet.createPacket(Packet64FrameType.TYPE_DATA, "{}".getBytes(), (short) 1);
messageQueue.put(msg);

package fj;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;

public class CountTask extends RecursiveTask {
    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
    protected Integer compute() {
        int sum = 0;
        //如果任务足够小就计算任务
        boolean canCompute = (end-start) <= THRESHOLD;
        if(canCompute) {
            for(int i = start; i <= end; i++) {
                sum += i;
            }
        }else{
            //如果任务大于阀值，就分裂成两个子任务计算
            int middle = (start + end) / 2;
            CountTask leftTask = new CountTask(start, middle);
            CountTask rightTask = new CountTask(middle + 1,end);
            //执行子任务
            leftTask.fork();
            rightTask.fork();
            //等待子任务执行完，并得到其结果
            int leftResult=leftTask.join();
            int rightResult=rightTask.join();
            //合并子任务
            sum = leftResult  + rightResult;
        }
        return sum;
    }

    public static void main(String[] args) {
        ForkJoinPool forkJoinPool =new ForkJoinPool();
        //生成一个计算任务，负责计算1+2+3+4
        CountTask task =new CountTask(1, 4);
        //执行一个任务
        Future result = forkJoinPool.submit(task);
        try{
            System.out.println(result.get());
        }catch(InterruptedException e) {

        }catch(ExecutionException e) {

        }
    }
}

if(task.isCompletedAbnormally())
{
    System.out.println(task.getException());
}

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

final void pushTask(ForkJoinTask t) {
    ForkJoinTask[] q; int s, m;
    if ((q = queue) != null) {    // ignore if queue removed
        long u = (((s = queueTop) & (m = q.length - 1)) << ASHIFT) + ABASE;
        UNSAFE.putOrderedObject(q, u, t);
        queueTop = s + 1;         // or use putOrderedInt
        if ((s -= queueBase) <= 2)
            pool.signalWork();
        else if (s == m)
            growQueue();
    }
}

public final V join() {
    if (doJoin() != NORMAL)
        return reportResult();
    else
        return getRawResult();
}

private V reportResult() {
    int s; Throwable ex;
    if ((s = status) == CANCELLED)
        throw new CancellationException();

    if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
        UNSAFE.throwException(ex);
    return getRawResult();
}

private int doJoin() {
    Thread t; ForkJoinWorkerThread w; int s; boolean completed;
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
        if ((s = status) < 0)
            return s;
        if ((w = (ForkJoinWorkerThread)t).unpushTask(this)) {
            try {
                completed = exec();
            } catch (Throwable rex) {
                return setExceptionalCompletion(rex);
            }

            if (completed)
                return setCompletion(NORMAL);
        }

        return w.joinTask(this);
    } else
        return externalAwaitDone();
}

ArrayBlockingQueue fairQueue = new  ArrayBlockingQueue(1000,true);

public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();

        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
}

public int compareTo(Delayed other) {
    if (other == this) // compare zero ONLY if same object
        return 0;

    if (other instanceof ScheduledFutureTask) {
        ScheduledFutureTask x = (ScheduledFutureTask)other;
        long diff = time - x.time;
        if (diff < 0)
            return -1;
        else if (diff > 0)
            return 1;
        else if (sequenceNumber < x.sequenceNumber)
            return -1;
        else
            return 1;
    }
    long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS));
    return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
}

ScheduledFutureTask(Runnable r, V result, long ns, long period) {
    super(r, result);
    this.time = ns;
    this.period = period;
    this.sequenceNumber = sequencer.getAndIncrement();
}

public long getDelay(TimeUnit unit) {
    return unit.convert(time - now(), TimeUnit.NANOSECONDS);
}

long delay = first.getDelay(TimeUnit.NANOSECONDS);

if (delay <= 0)
    return q.poll();
else if (leader != null)
    available.await();

Node pred = tryAppend(s, haveData);
return awaitMatch(s, pred, e, (how == TIMED), nanos);

private final Condition notFull;
private final Condition notEmpty;

public ArrayBlockingQueue(int capacity, boolean fair) {
    //省略其他代码
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();

    try {
        while (count == items.length)
            notFull.await();

        insert(e);
    } finally {
        lock.unlock();
    }
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();

    try {
        while (count == 0)
            notEmpty.await();

        return extract();
    } finally {
        lock.unlock();
    }
}

private void insert(E x) {
    items[putIndex] = x;
    putIndex = inc(putIndex);
    ++count;
    notEmpty.signal();
}

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();

    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    int interruptMode = 0;

    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);

        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }

    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;

    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();

    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);

}

public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    unsafe.park(false, 0L);
    setBlocker(t, null);
}

public native void park(boolean isAbsolute, long time);

void os::PlatformEvent::park() {
    int v ;

    for (;;) {
        v = _Event ;

        if (Atomic::cmpxchg (v-1, &_Event, v) == v) break ;
    }

    guarantee (v >= 0, "invariant") ;

    if (v == 0) {
        // Do this the hard way by blocking ...
        int status = pthread_mutex_lock(_mutex);
        assert_status(status == 0, status, "mutex_lock");
        guarantee (_nParked == 0, "invariant") ;
        ++ _nParked ;

        while (_Event < 0) {
            status = pthread_cond_wait(_cond, _mutex);
            // for some reason, under 2.7 lwp_cond_wait() may return ETIME ...
            // Treat this the same as if the wait was interrupted
            if (status == ETIME) { status = EINTR; }
                assert_status(status == 0 || status == EINTR, status, "cond_wait");
        }

          -- _nParked ;

        // In theory we could move the ST of 0 into _Event past the unlock(),
        // but then we'd need a MEMBAR after the ST.
        _Event = 0 ;
        status = pthread_mutex_unlock(_mutex);
        assert_status(status == 0, status, "mutex_unlock");
    }

    guarantee (_Event >= 0, "invariant") ;
}

"main" prio=5 tid=0x00007fc83c000000 nid=0x10164e000 waiting on condition [0x000000010164d000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x0000000140559fe8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
        at java.util.concurrent.ArrayBlockingQueue.put(ArrayBlockingQueue.java:324)
        at blockingqueue.ArrayBlockingQueueTest.main(ArrayBlockingQueueTest.java:11)

private transient volatile Node<e> tail = head;

public boolean offer(E e) {

    if (e == null) throw new NullPointerException();

    //入队前，创建一个入队节点
    Node</e><e> n = new Node</e><e>(e);
    retry:

    //死循环，入队不成功反复入队。
    for (;;) {
        //创建一个指向tail节点的引用
        Node</e><e> t = tail;
        //p用来表示队列的尾节点，默认情况下等于tail节点。
        Node</e><e> p = t;

        for (int hops = 0; ; hops++) {
        //获得p节点的下一个节点。
            Node</e><e> next = succ(p);

            //next节点不为空，说明p不是尾节点，需要更新p后在将它指向next节点
            if (next != null) {
                //循环了两次及其以上，并且当前节点还是不等于尾节点
                if (hops > HOPS && t != tail)
                    continue retry;

                p = next;
            }
            //如果p是尾节点，则设置p节点的next节点为入队节点。
            else if (p.casNext(null, n)) {
              //如果tail节点有大于等于1个next节点，则将入队节点设置成tair节点，更新失败了也没关系，因为失败了表示有其他线程成功更新了tair节点。

                if (hops >= HOPS)
                    casTail(t, n); // 更新tail节点，允许失败

                return true;
            }
            // p有next节点,表示p的next节点是尾节点，则重新设置p节点
            else {
                p = succ(p);
            }
        }
    }
}

final Node</e><e> succ(Node</e><e> p) {
    Node</e><e> next = p.getNext();
    return (p == next) ? head : next;
}

public boolean offer(E e) {

    if (e == null)
        throw new NullPointerException();

    Node</e><e> n = new Node</e><e>(e);

    for (;;) {
        Node</e><e> t = tail;

        if (t.casNext(null, n) && casTail(t, n)) {
            return true;
        }
    }
}

private static final int HOPS = 1;

public E poll() {
    Node</e><e> h = head;
    // p表示头节点，需要出队的节点
    Node</e><e> p = h;

    for (int hops = 0;; hops++) {
        // 获取p节点的元素
        E item = p.getItem();
        // 如果p节点的元素不为空，使用CAS设置p节点引用的元素为null,如果成功则返回p节点的元素。
        if (item != null && p.casItem(item, null)) {
            if (hops >= HOPS) {
                //将p节点下一个节点设置成head节点
                Node</e><e> q = p.getNext();
                updateHead(h, (q != null) ? q : p);
            }
            return item;
        }

        // 如果头节点的元素为空或头节点发生了变化，这说明头节点已经被另外一个线程修改了。那么获取p节点的下一个节点
        Node</e><e> next = succ(p);
        // 如果p的下一个节点也为空，说明这个队列已经空了
        if (next == null) {
            // 更新头节点。
            updateHead(h, p);
            break;
        }

        // 如果下一个元素不为空，则将头节点的下一个节点设置成头节点
        p = next;
    }
    return null;
}