C3P0:java.lang.InterruptedException: sleep interrupted

好好的就报异常了～～

警告: com.mchange.v2.resourcepool.BasicResourcePool@8fce95 -- Thread unexpectedly interrupted while performing an acquisition attempt.
java.lang.InterruptedException: sleep interrupted
	at java.lang.Thread.sleep(Native Method)
	at com.mchange.v2.resourcepool.BasicResourcePool$AcquireTask.run(BasicResourcePool.java:1805)
	at com.mchange.v2.async.ThreadPoolAsynchronousRunner$PoolThread.run(ThreadPoolAsynchronousRunner.java:547)

Disruptor (5) - 消费方式 EventHandler & WorkHandler

handleEventsWith & handleEventsWithWorkerPool

在 Disruptor.start方法执行之前，需要指定消息的处理方式。消费者对于同一消息： 都分别独立消费应当实现EventHandler接口；不重复消费则应当实现WorkHandler接口。

有不同的实现逻辑来支持不同的模式：

com.lmax.disruptor.dsl.ConsumerInfo

com.lmax.disruptor.EventProcessor

1. disruptor.handleEventsWith(EventHandler... handlers)：将多个EventHandler的实现类传入，封装成一个com.lmax.disruptor.dsl.EventHandlerGroup<T>。每个消费者都会对同一条消息进行独立消费，各个消费者之间不存在竞争。事件的处理类是BatchEventProcessor + EventProcessorInfo。
2. disruptor.handleEventsWithWorkerPool(WorkHandler... handlers)：将多个WorkHandler的实现类传入，封装成一个EventHandlerGroup<T>。对于同一条消息不重复消费；如果c0消费了消息m，则c1不再消费消息m。事件信息的处理类是com.lmax.disruptor.WorkerPool（N个WorkProcessor）+ WorkerPoolInfo。

Demo

public class CompareTest {
	public static int THREAD = 2; // 线程数量
	public static int PER = 1; // 单个线程生产数量
	public static int TOTAL_COUNT = THREAD * PER; // 数据总量
	public static int SIZE = 4; // 最大容量

	public static void main(String[] args) {
		println("线程数：" + THREAD + " 单线程生产量: " + PER + " 容量：" + SIZE + " 数据总量：" + TOTAL_COUNT);
		 DisruptorTest.execute();
	}

	public static void println(String msg) {
		System.out.println(DateTimeFormatter.ISO_INSTANT.format(Instant.now()) + "[" + Thread.currentThread().getName() + "] " + msg);
	}
}

public class DataWorkHandler implements WorkHandler<DataEvent> {
	public AtomicLong count = new AtomicLong(0);
	public String name = null;

	public DataWorkHandler(String name) {
		this.name = name;
	}

	@Override
	public void onEvent(DataEvent event) throws Exception {
		Thread.sleep(name.contentEquals("dataEventHandler1") ? 1000 : 100);
		CompareTest.println("handlerName: " + name + " 处理的sequence：" + event.getSequence()
				+ " count：" + count.incrementAndGet() + "  Disruptor 总耗时："
				+ (System.currentTimeMillis() - event.getStartTime()));
	}
}

• handleEventsWith同一消息被不同handler独立消费。 此时handler处理是无序的。

disruptor.handleEventsWith(new DataEventHandler("dataEventHandler1"), new DataEventHandler("dataEventHandler2"));

• 依赖串行.  对于同一消息前handler处理完结，后handler才处理。

disruptor.handleEventsWith(new DataEventHandler("dataEventHandler1")).then(new DataEventHandler("dataEventHandler2")).then(new DataEventHandler("dataEventHandler3"));

• handleEventsWithWorkerPool不重复消费。 

disruptor.handleEventsWithWorkerPool(new DataWorkHandler("dataWorkHandler1"), new DataWorkHandler("dataWorkHandler2"));

• 组合使用

disruptor.handleEventsWithWorkerPool(new DataWorkHandler("dataWorkHandler3"),new DataWorkHandler("dataWorkHandler4")).then(new DataEventHandler("dataEventHandler1"), new DataEventHandler("dataEventHandler2"));	

disruptor.handleEventsWith(new DataEventHandler("dataEventHandler1"), new DataEventHandler("dataEventHandler2")).thenHandleEventsWithWorkerPool(new DataWorkHandler("dataWorkHandler3"),new DataWorkHandler("dataWorkHandler4"));	

HanderThread 的 quit 和 interrupt

使用 Java 的时候，我们一般都是使用 interrupt 结束线程。那么使用 Android 的 HandlerThread 怎么结束一个线程呢？

我们知道 HanderThread 其实就是一个里吗有个 Looper 在死循环的普通 Thread

 @Override
    public void run() {
        mTid = Process.myTid();
        Looper.prepare();
        synchronized (this) {
            mLooper = Looper.myLooper();
            notifyAll();
        }
        Process.setThreadPriority(mPriority);
        onLooperPrepared();
        Looper.loop();
        mTid = -1;
    }

Looper.loop () 当然就是那个死循环了，他什么时候退出呢？

public static void loop() {
        final Looper me = myLooper();
        if (me == null) {
            throw new RuntimeException("No Looper; Looper.prepare() wasn''t called on this thread.");
        }
        final MessageQueue queue = me.mQueue;

        // Make sure the identity of this thread is that of the local process,
        // and keep track of what that identity token actually is.
        Binder.clearCallingIdentity();
        final long ident = Binder.clearCallingIdentity();

        for (;;) {
            Message msg = queue.next(); // might block
            if (msg == null) {
                // No message indicates that the message queue is quitting. 这里退出
                return;
            }

            // This must be in a local variable, in case a UI event sets the logger
            Printer logging = me.mLogging;
            if (logging != null) {
                logging.println(">>>>> Dispatching to " + msg.target + " " +
                        msg.callback + ": " + msg.what);
            }

            msg.target.dispatchMessage(msg);

            if (logging != null) {
                logging.println("<<<<< Finished to " + msg.target + " " + msg.callback);
            }

            // Make sure that during the course of dispatching the
            // identity of the thread wasn''t corrupted.
            final long newIdent = Binder.clearCallingIdentity();
            if (ident != newIdent) {
                Log.wtf(TAG, "Thread identity changed from 0x"
                        + Long.toHexString(ident) + " to 0x"
                        + Long.toHexString(newIdent) + " while dispatching to "
                        + msg.target.getClass().getName() + " "
                        + msg.callback + " what=" + msg.what);
            }

            msg.recycleUnchecked();
        }
    }

原来是

 Message msg = queue.next(); // might block
            if (msg == null) {
                // No message indicates that the message queue is quitting.
                return;
            }

那什么时候 msg = null 呢？这里的 queue 是 MessageQueue.java

Message next() {
        // Return here if the message loop has already quit and been disposed.
        // This can happen if the application tries to restart a looper after quit
        // which is not supported.
        final long ptr = mPtr;
        if (ptr == 0) {
            return null;
        }

        int pendingIdleHandlerCount = -1; // -1 only during first iteration
        int nextPollTimeoutMillis = 0;
        for (;;) {
            if (nextPollTimeoutMillis != 0) {
                Binder.flushPendingCommands();
            }

            nativePollOnce(ptr, nextPollTimeoutMillis);

            synchronized (this) {
                // Try to retrieve the next message.  Return if found.
                final long now = SystemClock.uptimeMillis();
                Message prevMsg = null;
                Message msg = mMessages;
                if (msg != null && msg.target == null) {
                    // Stalled by a barrier.  Find the next asynchronous message in the queue.
                    do {
                        prevMsg = msg;
                        msg = msg.next;
                    } while (msg != null && !msg.isAsynchronous());
                }
                if (msg != null) {
                    if (now < msg.when) {
                        // Next message is not ready.  Set a timeout to wake up when it is ready.
                        nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE);
                    } else {
                        // Got a message.
                        mBlocked = false;
                        if (prevMsg != null) {
                            prevMsg.next = msg.next;
                        } else {
                            mMessages = msg.next;
                        }
                        msg.next = null;
                        if (false) Log.v("MessageQueue", "Returning message: " + msg);
                        return msg;
                    }
                } else {
                    // No more messages.
                    nextPollTimeoutMillis = -1;
                }

                // Process the quit message now that all pending messages have been handled.
                if (mQuitting) {
                    dispose();
                    return null;
                }

                // If first time idle, then get the number of idlers to run.
                // Idle handles only run if the queue is empty or if the first message
                // in the queue (possibly a barrier) is due to be handled in the future.
                if (pendingIdleHandlerCount < 0
                        && (mMessages == null || now < mMessages.when)) {
                    pendingIdleHandlerCount = mIdleHandlers.size();
                }
                if (pendingIdleHandlerCount <= 0) {
                    // No idle handlers to run.  Loop and wait some more.
                    mBlocked = true;
                    continue;
                }

                if (mPendingIdleHandlers == null) {
                    mPendingIdleHandlers = new IdleHandler[Math.max(pendingIdleHandlerCount, 4)];
                }
                mPendingIdleHandlers = mIdleHandlers.toArray(mPendingIdleHandlers);
            }

            // Run the idle handlers.
            // We only ever reach this code block during the first iteration.
            for (int i = 0; i < pendingIdleHandlerCount; i++) {
                final IdleHandler idler = mPendingIdleHandlers[i];
                mPendingIdleHandlers[i] = null; // release the reference to the handler

                boolean keep = false;
                try {
                    keep = idler.queueIdle();
                } catch (Throwable t) {
                    Log.wtf("MessageQueue", "IdleHandler threw exception", t);
                }

                if (!keep) {
                    synchronized (this) {
                        mIdleHandlers.remove(idler);
                    }
                }
            }

            // Reset the idle handler count to 0 so we do not run them again.
            pendingIdleHandlerCount = 0;

            // While calling an idle handler, a new message could have been delivered
            // so go back and look again for a pending message without waiting.
            nextPollTimeoutMillis = 0;
        }
    }

有两个地方，第一个是初始化 C++ 层对象的一个指针。第二个就是 mQuiting =true

调用 quit 的时候会导致 mQuiting = true, 那 Looper 就会退出了.HanderThread 也就退出了。此时也把 C++ 层真正的 NativeMessageQueue 对象销毁了

而 interrupt 其实是调用了 Thread.java 的 interrupt 方法，也就是 把暴力线程终止了.

PS: 为了提高效率，MessageQueue.java 不负责真正的队列操作， 真正的对了操作使用了 linux pipe 实现， 所以有了那个 mPtr, 它指向 Linux 层的 NativeMessageQueue. 更多请参考 http://blog.csdn.net/u010018855/article/details/50221079

今天关于Interrupt Handler使用哪个堆栈 – Linux和狸奴小睡不知愁,忙添落花作锦衾的介绍到此结束，谢谢您的阅读，有关android中的runOnMainThread vs Handler-何时针对上述场景使用哪个、C3P0:java.lang.InterruptedException: sleep interrupted、Disruptor (5) - 消费方式 EventHandler & WorkHandler、HanderThread 的 quit 和 interrupt等更多相关知识的信息可以在本站进行查询。