优化消费者池构造、消费者获取策略

Author: seaswalker

src/main/java/consumer/pool/ConsumerPool.java

@@ -5,47 +5,21 @@
 import consumer.lifecycle.LifeCycle;
 
 /**
- * 消费者池，一个池中包含下列元素:
- * <ul>
- * <li>一定数量的单线程消费者(SPSC).</li>
- * <li>一个多线程消费者(MPMC).</li>
- * </ul>
+ * 消费者池.
  *
  * @author skywalker
  */
 public interface ConsumerPool<T> extends LifeCycle {
 
     /**
-     * 申请一个消费者{@link Consumer}，遵循如下的策略:
-     * <ul>
-     * <li>如果当前尚有单线程消费者存在，返回之.</li>
-     * <li>返回多线程消费者.</li>
-     * </ul>
+     * 申请一个消费者{@link Consumer}.
      *
      * @return {@linkplain Consumer} 可用的消费者
      */
     Consumer<T> acquire();
 
-    /**
-     * 申请单线程消费者(SPSC).
-     *
-     * @return {@link Consumer} 如果当前没有可用的消费者，返回null
-     * @see #acquire()
-     */
-    Consumer<T> acquireSPSC();
-
-    /**
-     * 申请多线程消费者(MPSC).
-     *
-     * @return {@link Consumer} 如果当前没有可用的消费者，返回null
-     * @see #acquire()
-     */
-    MultiThreadsConsumer<T> acquireMPMC();
-
     /**
      * 释放{@link Consumer}.当一个生产者不再使用一个消费者时，务必确保此方法被调用.
-     *
-     * @param consumer {@linkplain Consumer}
      */
     void release(Consumer<T> consumer);
 

src/main/java/consumer/pool/DefaultConsumerPool.java

@@ -1,12 +1,16 @@
 package consumer.pool;
 
 import consumer.Consumer;
-import consumer.MultiThreadsConsumer;
 import consumer.cas.AbstractSPCASConsumer;
 import consumer.cas.AbstractMPCASConsumer;
 import consumer.cas.strategy.RetryStrategy;
 import consumer.lifecycle.LifeCycle;
 import consumer.lifecycle.StateCheckDelegate;
+import consumer.pool.dispatch.DispatchStrategy;
+import consumer.pool.dispatch.RoundRobinStrategy;
+import consumer.pool.dispatch.SingleProducerStrategy;
+import consumer.pool.internal.InternalMPConsumer;
+import consumer.pool.internal.InternalSPConsumer;
 import org.slf4j.Logger;
 import consumer.util.Util;
 
@@ -24,29 +28,14 @@
 public class DefaultConsumerPool<T> implements ConsumerPool<T> {
 
     /**
-     * 每个单生产者队列消费线程数量.
+     * 每个消费者的消费线程数.
      */
-    private final int spThreads;
-    /**
-     * 单生产者队列大小.
-     */
-    private final int spQueueSize;
-    /**
-     * 多生产者消费线程数.
-     */
-    private final int mpThreads;
-    /**
-     * 多生产者队列大小.
-     */
-    private final int mpQueueSize;
+    private final int threads;
+    private final int queueSize;
     private final ConsumeActionFactory<T> factory;
-    /**
-     * 消费者列表，采取如下的存储策略:
-     * <p>最后一个为多生产者模型，前面的为但生产者模型.</p>
-     */
-    private final List<ConsumerWrapper> list;
+    private final boolean isSingleProducer;
     private final StateCheckDelegate delegate = StateCheckDelegate.getInstance();
-    private final Object monitor = new Object();
+    private final List<Consumer<T>> list;
     /**
      * 消费者总数.
      */
@@ -56,6 +45,7 @@ public class DefaultConsumerPool<T> implements ConsumerPool<T> {
     private RetryStrategy<T> retryStrategy;
     private ThreadNameGenerator threadNameGenerator;
     private Thread.UncaughtExceptionHandler handler;
+    private DispatchStrategy<T> dispatchStrategy;
 
     private volatile State state = State.INIT;
 
@@ -64,36 +54,47 @@ public class DefaultConsumerPool<T> implements ConsumerPool<T> {
      */
     private static final String defaultTerminateThreadName = "DefaultConsumerPool-terminate-thread";
 
-    public DefaultConsumerPool(int sp, int spThreads, int mpThreads, int spQueueSize, int mpQueueSize, ConsumeActionFactory<T> factory) {
+    public DefaultConsumerPool(boolean isSingleProducer, int number, int consumerThreads, int queueSize, ConsumeActionFactory<T> factory) {
         Objects.requireNonNull(factory);
-        this.mpThreads = mpThreads;
-        this.spThreads = spThreads;
         this.factory = factory;
-        this.spQueueSize = spQueueSize;
-        this.mpQueueSize = mpQueueSize;
-        this.consumers = sp + 1;
-        this.list = new ArrayList<>(this.consumers);
+        this.consumers = number;
+        this.queueSize = queueSize;
+        this.threads = consumerThreads;
+        this.isSingleProducer = isSingleProducer;
+        this.list = new ArrayList<>(consumers);
     }
 
     @Override
     public boolean start() {
         delegate.checkStart(this);
-        for (int i = 0; i < consumers - 1; i++) {
-            InternalSPConsumer spConsumer = new InternalSPConsumer(spQueueSize, spThreads);
-            prepareConsumer(spConsumer);
-            if (!spConsumer.start()) {
+        if (isSingleProducer) {
+            //单生产者模式，我们需要number - 1个InternalSPConsumer和1个InternalMPConsumer
+            for (int i = 0; i < consumers - 1; i++) {
+                InternalSPConsumer<T> spConsumer = new InternalSPConsumer<>(queueSize, threads, factory, threadNameGenerator);
+                prepareConsumer(spConsumer);
+                if (!spConsumer.start()) {
+                    return false;
+                }
+                list.add(spConsumer);
+            }
+            InternalMPConsumer<T> mpConsumer = new InternalMPConsumer<>(queueSize, threads, factory, threadNameGenerator);
+            prepareConsumer(mpConsumer);
+            if (!mpConsumer.start()) {
                 return false;
             }
-            ConsumerWrapper wrapper = new ConsumerWrapper(spConsumer);
-            spConsumer.wrapper = wrapper;
-            list.add(wrapper);
-        }
-        AbstractMPCASConsumer<T> mpConsumer = new InternalMPConsumer(mpQueueSize, mpThreads);
-        prepareConsumer(mpConsumer);
-        if (!mpConsumer.start()) {
-            return false;
+            list.add(mpConsumer);
+            dispatchStrategy = new SingleProducerStrategy<>(list);
+        } else {
+            for (int i = 0; i < consumers; i++) {
+                AbstractMPCASConsumer<T> mpConsumer = new InternalMPConsumer<>(queueSize, threads, factory, threadNameGenerator);
+                prepareConsumer(mpConsumer);
+                if (!mpConsumer.start()) {
+                    return false;
+                }
+                list.add(mpConsumer);
+            }
+            dispatchStrategy = new RoundRobinStrategy<>(list);
         }
-        list.add(new ConsumerWrapper(mpConsumer));
         this.state = State.RUNNING;
         return true;
     }
@@ -132,7 +133,7 @@ private Future<Void> terminateHelper(Function<Consumer<T>, Future<Void>> functio
         delegate.checkTerminated(this);
         final Future[] futures = new Future[this.consumers];
         for (int i = 0; i < this.consumers; i++) {
-            futures[i] = function.apply(list.get(i).consumer);
+            futures[i] = function.apply(list.get(i));
         }
         final CompletableFuture<Void> future = new CompletableFuture<>();
         new Thread(() -> {
@@ -164,48 +165,12 @@ public State getState() {
     @Override
     public Consumer<T> acquire() {
         delegate.checkRunning(this);
-        Consumer<T> result = acquireSPSC();
-        if (result == null) {
-            if (log != null && log.isDebugEnabled()) {
-                log.debug("Currently there are no sp consumer available, so use mp instead.");
-            }
-            result = acquireMPMC();
-        }
-        return result;
-    }
-
-    @Override
-    public Consumer<T> acquireSPSC() {
-        delegate.checkRunning(this);
-        Consumer<T> result = null;
-        synchronized (monitor) {
-            for (int i = 0; i < consumers - 1; i++) {
-                ConsumerWrapper wrapper = list.get(i);
-                if (wrapper.available) {
-                    result = wrapper.consumer;
-                    wrapper.available = false;
-                    break;
-                }
-            }
-        }
-        return result;
-    }
-
-    @Override
-    public MultiThreadsConsumer<T> acquireMPMC() {
-        delegate.checkRunning(this);
-        return (MultiThreadsConsumer<T>) list.get(consumers - 1).consumer;
+        return dispatchStrategy.acquire();
     }
 
     @Override
     public void release(Consumer<T> consumer) {
-        if (!(consumer instanceof MultiThreadsConsumer)) {
-            //无需释放
-            synchronized (monitor) {
-                InternalSPConsumer internal = (InternalSPConsumer) consumer;
-                internal.wrapper.available = true;
-            }
-        }
+        dispatchStrategy.release(consumer);
     }
 
     @SuppressWarnings("unused")
@@ -249,68 +214,4 @@ public ThreadNameGenerator getThreadNameGenerator() {
         return threadNameGenerator;
     }
 
-    /**
-     * {@link AbstractSPCASConsumer}实现，将其consume方法委托给{@link ConsumeAction#consume(Object)}.
-     */
-    private class InternalSPConsumer extends AbstractSPCASConsumer<T> {
-
-        private ConsumerWrapper wrapper;
-
-        private final ConsumeAction<T> action;
-
-        InternalSPConsumer(int queueSize, int threads) {
-            super(queueSize, threads);
-            this.action = factory.newAction();
-        }
-
-        @Override
-        public void consume(T task) {
-            action.consume(task);
-        }
-
-        @Override
-        protected String getThreadName(Thread t) {
-            return (threadNameGenerator == null ? super.getThreadName(t) : threadNameGenerator.generate(t));
-        }
-
-    }
-
-    /**
-     * {@link AbstractMPCASConsumer}实现，将其consume方法委托给{@link ConsumeAction#consume(Object)}.
-     */
-    private class InternalMPConsumer extends AbstractMPCASConsumer<T> {
-
-        private final ConsumeAction<T> action;
-
-        InternalMPConsumer(int queueSize, int threads) {
-            super(queueSize, threads);
-            this.action = factory.newAction();
-        }
-
-        @Override
-        public void consume(T task) {
-            action.consume(task);
-        }
-
-        @Override
-        protected String getThreadName(Thread t) {
-            return (threadNameGenerator == null ? super.getThreadName(t) : threadNameGenerator.generate(t));
-        }
-
-    }
-
-    /**
-     * 对{@link Consumer}进行包装，增加是否可用的标志位.
-     */
-    private class ConsumerWrapper {
-
-        private boolean available = true;
-        private final Consumer<T> consumer;
-
-        ConsumerWrapper(Consumer<T> consumer) {
-            this.consumer = consumer;
-        }
-
-    }
-
 }

src/main/java/consumer/pool/dispatch/AbstractDispatchStrategy.java

@@ -0,0 +1,42 @@
+package consumer.pool.dispatch;
+
+import consumer.Consumer;
+import consumer.pool.dispatch.DispatchStrategy;
+
+import java.util.List;
+import java.util.concurrent.Future;
+
+/**
+ * 屏蔽掉{@link consumer.lifecycle.LifeCycle}的相关方法.
+ *
+ * @author skywalker
+ */
+public abstract class AbstractDispatchStrategy<T> implements DispatchStrategy<T> {
+
+    protected final List<Consumer<T>> consumers;
+
+    protected AbstractDispatchStrategy(List<Consumer<T>> consumers) {
+        this.consumers = consumers;
+    }
+
+    @Override
+    public final boolean start() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public final Future<Void> terminate() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public final Future<Void> terminateNow() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public final State getState() {
+        throw new UnsupportedOperationException();
+    }
+
+}

src/main/java/consumer/pool/dispatch/DispatchStrategy.java

@@ -0,0 +1,11 @@
+package consumer.pool.dispatch;
+
+import consumer.pool.ConsumerPool;
+
+/**
+ * 消费者的分配策略.
+ *
+ * @author skywalker
+ */
+public interface DispatchStrategy<T> extends ConsumerPool<T> {
+}