添加配置可选项，是选择一次性消费，还是幂等性消费，

是先提交offset 还是后提交offset.

src/Kafka/Consumer/Process.php

@@ -14,6 +14,8 @@
 
 namespace Kafka\Consumer;
 
+use Kafka\ConsumerConfig;
+
 /**
 +------------------------------------------------------------------------------
 * Kafka protocol since Kafka v0.8
@@ -660,7 +662,7 @@ public function succFetch($result, $fd)
     // }}}
     // {{{ protected function commit()
 
-    protected function commit()
+    protected function consume_msg()
     {
         foreach ($this->messages as $topic => $value) {
             foreach ($value as $part => $messages) {
@@ -673,7 +675,18 @@ protected function commit()
         }
 
         $this->messages = array();
-
+    }
+
+
+    protected function commit()
+    {
+        $config= ConsumerConfig::getInstance();
+        if($config->getConsumeMode() == ConsumerConfig::CONSUME_BEFORE_COMMIT_OFFSET)
+        {
+            $this->consume_msg();
+        }
+
+
         $broker = \Kafka\Broker::getInstance();
         $groupBrokerId = $broker->getGroupBrokerId();
         $connect = $broker->getMetaConnect($groupBrokerId);
@@ -732,6 +745,10 @@ public function succCommit($result)
                 }
             }
         }
+        if(ConsumerConfig::getInstance()->getConsumeMode() == ConsumerConfig::CONSUME_AFTER_COMMIT_OFFSET)
+        {
+            $this->consume_msg();
+        }
     }
 
     // }}}

src/Kafka/ConsumerConfig.php

@@ -130,4 +130,20 @@ public function setTopics($topics)
 
     // }}}
     // }}}
+
+    protected $runtime_options = [
+        'consume_mode' => self::CONSUME_AFTER_COMMIT_OFFSET
+    ];
+    const CONSUME_AFTER_COMMIT_OFFSET = 1;
+    const CONSUME_BEFORE_COMMIT_OFFSET = 2;
+
+    public function setConsumeMode($mode)
+    {
+        $this->runtime_options['consume_mode'] = $mode;
+    }
+
+    public function getConsumeMode()
+    {
+        return $this->runtime_options['consume_mode'];
+    }
 }