改名

2020-10-28 16:49:44 +08:00
parent b4e4cc8ac3
commit 07f1d8e650
1 changed files with 3 additions and 2 deletions
@@ -104,7 +104,6 @@ class Producer extends Component
 		if ($rk->getOutQLen() > 0) {
 			$this->clean($rk, $timeout);
 		}
-
 		$topic = $rk->newTopic($this->_topic, $this->topicConf);
 		$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
 		$this->clean($rk, $timeout);
@@ -117,7 +116,9 @@ class Producer extends Component
 	 */
 	private function clean($rk, $timeout)
 	{
-		$rk->poll(0);
+		for ($length = 0; $length < $rk->getOutQLen(); $length++) {
+			$rk->poll(0);
+		}
 		$rk->flush($timeout);
 	}