优化ConsumerPool模型

Author: seaswalker

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>skywalker</groupId>
     <artifactId>Consumer</artifactId>
-    <version>1.2-SNAPSHOT</version>
+    <version>1.3-SNAPSHOT</version>
     <build>
         <plugins>
             <plugin>
@@ -43,7 +43,7 @@
         <dependency>
             <groupId>org.jctools</groupId>
             <artifactId>jctools-core</artifactId>
-            <version>2.0</version>
+            <version>2.0.1</version>
         </dependency>
 
         <dependency>

src/main/java/consumer/cas/AbstractMPCASConsumer.java

@@ -0,0 +1,24 @@
+package consumer.cas;
+
+import consumer.MultiThreadsConsumer;
+import consumer.queue.SQueue;
+import consumer.queue.cas.MpmcBasedQueue;
+import consumer.queue.cas.MpscBasedQueue;
+
+/**
+ * {@link MultiThreadsConsumer}实现，暂且只提供给予CAS的多线程消费者实现.
+ *
+ * @author skywalker
+ */
+public abstract class AbstractMPCASConsumer<T> extends AbstractSPCASConsumer<T> implements MultiThreadsConsumer<T> {
+
+    public AbstractMPCASConsumer(int queueSize, int threads) {
+        super(queueSize, threads);
+    }
+
+    @Override
+    protected final SQueue<T> newQueue() {
+        return (threads > 1 ? new MpmcBasedQueue<T>(queueSize) : new MpscBasedQueue<T>(queueSize));
+    }
+
+}

src/main/java/consumer/cas/AbstractMultiThreadsConsumer.java

@@ -4,28 +4,40 @@
 import consumer.cas.strategy.BlockStrategy;
 import consumer.cas.strategy.RetryStrategy;
 import consumer.queue.SQueue;
+import consumer.queue.cas.SpmcBasedQueue;
 import consumer.queue.cas.SpscBasedQueue;
 
 import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 
 /**
- * {@link AbstractQueuedConsumer}骨架实现，基于CAS操作的消费者实现.
+ * {@link AbstractQueuedConsumer}骨架实现，基于无锁队列实现，子类可指定消费者线程数，
+ * 如果有一个消费线程，那么使用{@link org.jctools.queues.SpscArrayQueue}，否则使用
+ * {@link org.jctools.queues.SpmcArrayQueue}.
  * <br>
  * 默认采用阻塞的等待策略.
  *
  * @author skywalker
  */
-public abstract class AbstractCASConsumer<T> extends AbstractQueuedConsumer<T> {
+public abstract class AbstractSPCASConsumer<T> extends AbstractQueuedConsumer<T> {
 
     private RetryStrategy<T> retryStrategy = new BlockStrategy<T>();
 
-    public AbstractCASConsumer(int queueSize) {
+    /**
+     * 消费线程的数量.
+     */
+    protected final int threads;
+
+    public AbstractSPCASConsumer(int queueSize, int threads) {
         super(queueSize);
+        this.threads = threads;
     }
 
     @Override
     protected SQueue<T> newQueue() {
-        return new SpscBasedQueue<T>(queueSize);
+        return (threads > 1 ? new SpmcBasedQueue<T>(queueSize) : new SpscBasedQueue<T>(queueSize));
     }
 
     @Override
@@ -59,4 +71,14 @@ public void setRetryStrategy(RetryStrategy<T> retryStrategy) {
         this.retryStrategy = retryStrategy;
     }
 
+    @Override
+    protected final ExecutorService startExecutor(ThreadFactory threadFactory) {
+        ExecutorService service = Executors.newFixedThreadPool(threads, threadFactory);
+        int index = 0;
+        while (index++ < threads) {
+            service.execute(this);
+        }
+        return service;
+    }
+
 }

…va/consumer/cas/AbstractCASConsumer.java …/consumer/cas/AbstractSPCASConsumer.javasrc/main/java/consumer/cas/AbstractCASConsumer.java renamed to src/main/java/consumer/cas/AbstractSPCASConsumer.java src/main/java/consumer/cas/AbstractCASConsumer.java renamed to src/main/java/consumer/cas/AbstractSPCASConsumer.java

@@ -1,9 +1,10 @@
 package consumer.cas.strategy;
 
+import consumer.cas.AbstractSPCASConsumer;
 import consumer.queue.SQueue;
 
 /**
- * 当{@link consumer.cas.AbstractCASConsumer}获取任务失败时采取的重试策略.
+ * 当{@link AbstractSPCASConsumer}获取任务失败时采取的重试策略.
  *
  * @author skywalker
  */