diff --git a/HttpServer/Server.php b/HttpServer/Server.php
index 3dd55810..99249bcb 100644
--- a/HttpServer/Server.php
+++ b/HttpServer/Server.php
@@ -14,13 +14,9 @@ use HttpServer\Service\Receive;
 use HttpServer\Service\Packet;
 use HttpServer\Service\Websocket;
 use Exception;
-use JetBrains\PhpStorm\Pure;
 use ReflectionException;
-use Rpc\Service;
 use Snowflake\Abstracts\Config;
 use Snowflake\Error\LoggerProcess;
-use Snowflake\Event;
-use Snowflake\Exception\ComponentException;
 use Snowflake\Exception\ConfigException;
 use Snowflake\Exception\NotFindClassException;
 use Snowflake\Process\Biomonitoring;
@@ -53,7 +49,6 @@ class Server extends HttpService
     const PACKAGE = 'PACKAGE';
     const WEBSOCKET = 'WEBSOCKET';
 
-    private array $listening = [];
     private array $server = [
         'HTTP'      => [SWOOLE_TCP, Http::class],
         'TCP'       => [SWOOLE_TCP, Receive::class],
@@ -225,10 +220,10 @@ class Server extends HttpService
      * @throws ConfigException
      * @throws Exception
      */
-    public function onProcessListener(): void
+    public function onProcessListener(): mixed
     {
         if (!($this->swoole instanceof \Swoole\Server)) {
-            return;
+            return $this->swoole;
         }
 
         $processes = Config::get('processes');
@@ -237,6 +232,7 @@ class Server extends HttpService
         } else {
             $this->deliveryProcess($this->process);
         }
+        return $this->swoole;
     }
 
 
@@ -359,12 +355,32 @@ class Server extends HttpService
         if (isset($config['settings']) && is_array($config['settings'])) {
             $newListener->set($config['settings']);
         }
-        $this->onListenerBind($config, $this->swoole);
+        $this->onListenerBind($config);
 
         return $this->swoole;
     }
 
 
+    /**
+     * @param $mode
+     * @return bool
+     */
+    public static function isTcp($mode)
+    {
+        return in_array($mode, [SWOOLE_SOCK_TCP, SWOOLE_TCP, SWOOLE_TCP6, SWOOLE_SOCK_TCP6]);
+    }
+
+
+    /**
+     * @param $mode
+     * @return bool
+     */
+    public static function isUdp($mode)
+    {
+        return in_array($mode, [SWOOLE_SOCK_UDP, SWOOLE_UDP, SWOOLE_UDP6, SWOOLE_SOCK_UDP6]);
+    }
+
+
     /**
      * @return Packet|Websocket|Receive|Http|null
      * @throws ConfigException
@@ -372,12 +388,11 @@ class Server extends HttpService
      */
     private function startRpcService(): Packet|Websocket|Receive|Http|null
     {
-        $rpcService = Config::get('rpc.enable', []);
-        if ($rpcService === true) {
-            /** @var Service $service */
-            $service = Snowflake::app()->get('rpc-service');
-            $service->instance($this->swoole);
+        $rpcService = Config::get('rpc', []);
+        if (empty($rpcService)) {
+            return $this->swoole;
         }
+        $this->addListener($rpcService);
         return $this->swoole;
     }
 
@@ -394,17 +409,13 @@ class Server extends HttpService
         if (is_array($config['settings'] ?? null)) {
             $settings = array_merge($settings, $config['settings']);
         }
-        $this->swoole = $this->createServer($class, $config);
-        $settings['daemonize'] = $this->daemon;
-        if (!isset($settings['pid_file'])) {
-            $settings['pid_file'] = PID_PATH;
-        }
-
         $this->debug(Snowflake::listen($config));
-        $this->swoole->set($settings);
-        $this->onProcessListener();
-
-        return $this->swoole;
+        $this->swoole = $this->createServer($class, $config);
+        $this->swoole->set(array_merge($settings, [
+            'daemonize' => $this->daemon,
+            'pid_file'  => $settings['pid_file'] ?? PID_PATH
+        ]));
+        return $this->onProcessListener();
     }
 
 
@@ -419,16 +430,6 @@ class Server extends HttpService
     }
 
 
-    /**
-     * @param $listen
-     * @return bool
-     */
-    #[Pure] public function isListen($listen): bool
-    {
-        return in_array($listen, $this->listenTypes);
-    }
-
-
     /**
      * @param $config
      * @param $newListener
@@ -437,16 +438,10 @@ class Server extends HttpService
      * @throws ReflectionException
      * @throws Exception
      */
-    private function onListenerBind($config, $newListener): Packet|Websocket|Receive|Http|null
+    private function onListenerBind($config): Packet|Websocket|Receive|Http|null
     {
-        if (!in_array($config['type'], [self::HTTP, self::TCP, self::PACKAGE])) {
-            throw new Exception('Unknown server type(' . $config['type'] . ').');
-        }
-        if ($config['type'] == self::HTTP && !$this->swoole->getCallback('request')) {
-            $this->onBindCallback('request', [make(OnRequest::class), 'onHandler']);
-        } else {
-            $this->noHttp($newListener, $config);
-        }
+        $this->bindServerEvent($config['type']);
+
         $this->debug(sprintf('Check listen %s::%d -> ok', $config['host'], $config['port']));
         $this->listenTypes[] = $config['type'];
         return $this->swoole;
@@ -458,14 +453,20 @@ class Server extends HttpService
      * @param $config
      * @throws Exception
      */
-    private function noHttp($newListener, $config)
+    private function bindServerEvent($type = self::TCP)
     {
-        $this->onBindCallback('connect', [make(OnConnect::class), 'onHandler']);
-        $this->onBindCallback('close', [make(OnClose::class), 'onHandler']);
-        if ($config['type'] == self::TCP) {
-            $this->onBindCallback('receive', [make(OnReceive::class), 'onHandler']);
+        if (in_array($type, [self::PACKAGE, self::TCP])) {
+            $this->onBindCallback('connect', [make(OnConnect::class), 'onHandler']);
+            $this->onBindCallback('close', [make(OnClose::class), 'onHandler']);
+            if ($type == self::PACKAGE) {
+                $this->onBindCallback('packet', [make(OnPacket::class), 'onHandler']);
+            } else if ($type == self::TCP) {
+                $this->onBindCallback('receive', [make(OnReceive::class), 'onHandler']);
+            }
+        } else if ($type === self::HTTP) {
+            $this->onBindCallback('request', [make(OnRequest::class), 'onHandler']);
         } else {
-            $this->onBindCallback('packet', [make(OnPacket::class), 'onHandler']);
+            throw new Exception('Unknown server type(' . $type . ').');
         }
     }
 
diff --git a/Kafka/KafkaProviders.php b/Kafka/KafkaProviders.php
index ba0a720c..44a73cf7 100644
--- a/Kafka/KafkaProviders.php
+++ b/Kafka/KafkaProviders.php
@@ -36,7 +36,7 @@ class KafkaProviders extends Providers
 			return;
 		}
 
-		$kafkaServers = Config::get('kafka.servers', []);
+		$kafkaServers = Config::get('kafka.consumers', []);
 		if (empty($kafkaServers)) {
 			return;
 		}
diff --git a/Kafka/Producer.php b/Kafka/Producer.php
index 7ee183d4..7e6d7396 100644
--- a/Kafka/Producer.php
+++ b/Kafka/Producer.php
@@ -28,164 +28,206 @@ use Snowflake\Snowflake;
 class Producer extends Component
 {
 
-	private string $_topic = '';
+    private string $_topic = '';
 
 
-	private Conf $conf;
-	private TopicConf $topicConf;
+    private Conf $conf;
+    private TopicConf $topicConf;
 
-	private ?\RdKafka\Producer $producer = null;
+    private ?\RdKafka\Producer $producer = null;
+
+    private bool $isAck = true;
+
+    /**
+     * Producer constructor.
+     * @param array $config
+     * @throws Exception
+     */
+    public function __construct($config = [])
+    {
+        parent::__construct($config);
+        if (!class_exists(Conf::class)) {
+            return;
+        }
+        $this->conf = new Conf();
+        $this->topicConf = new TopicConf();
+        $this->conf->setErrorCb(function ($kafka, $err, $reason) {
+            $this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
+        });
+    }
 
 
-	/**
-	 * Producer constructor.
-	 * @param array $config
-	 * @throws Exception
-	 */
-	public function __construct($config = [])
-	{
-		parent::__construct($config);
-		if (!class_exists(Conf::class)) {
-			return;
-		}
-		$this->conf = new Conf();
-		$this->topicConf = new TopicConf();
-	}
+    /**
+     * @param $servers
+     * @return Producer
+     */
+    public function setBrokers(string $servers): static
+    {
+        $this->conf->set('metadata.broker.list', $servers);
+        return $this;
+    }
 
 
-	/**
-	 * @param $servers
-	 * @return Producer
-	 */
-	public function setBrokers(string $servers): static
-	{
-		$this->conf->set('metadata.broker.list', $servers);
-		return $this;
-	}
+    /**
+     * @param string $groupId
+     * @return Producer
+     */
+    public function setGroupId(string $groupId): static
+    {
+        $this->conf->set('group.id', $groupId);
+        return $this;
+    }
 
 
-	/**
-	 * @param string $groupId
-	 * @return Producer
-	 */
-	public function setGroupId(string $groupId): static
-	{
-		$this->conf->set('group.id', $groupId);
-		return $this;
-	}
+    /**
+     * @param $servers
+     * @return Producer
+     */
+    public function setTopic(string $servers): static
+    {
+        $this->_topic = $servers;
+        return $this;
+    }
 
 
-	/**
-	 * @param $servers
-	 * @return Producer
-	 */
-	public function setTopic(string $servers): static
-	{
-		$this->_topic = $servers;
-		return $this;
-	}
+    /**
+     * @param string $topic
+     * @param array $params
+     * @param string|null $groupId
+     * @throws Exception
+     */
+    public function dispatch(string $topic, array $params = [], string $groupId = null)
+    {
+        $this->beforePushMessage($topic, $groupId);
+        $this->sendMessage($topic, [$params]);
+    }
+
+    /**
+     * @return \RdKafka\Producer
+     * @throws Exception
+     */
+    private function getProducer(): \RdKafka\Producer
+    {
+        $pool = Snowflake::app()->getChannel();
+        return $pool->pop(\RdKafka\Producer::class, function () {
+            return Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]);
+        });
+    }
 
 
-	/**
-	 * @param string $topic
-	 * @param array $params
-	 * @param string|null $groupId
-	 * @throws Exception
-	 */
-	public function dispatch(string $topic, array $params = [], string $groupId = null)
-	{
-		$consumers = Config::get('kafka.consumers.' . $topic);
-		if (empty($consumers) || !is_array($consumers)) {
-			return;
-		}
-		if (!isset($consumers['brokers'])) {
-			throw new Exception('You need set brokers config.');
-		}
-		if (!empty($groupId)) {
-			$consumers['groupId'] = $groupId;
-		} else if (!isset($consumers['groupId'])) {
-			$consumers['groupId'] = $topic . ':' . Snowflake::localhost();
-		}
-		$this->setGroupId($consumers['groupId'])->setTopic($topic)
-			->setBrokers($consumers['brokers'])
-			->delivery(swoole_serialize($params));
-	}
+    /**
+     * @return \RdKafka\Producer
+     * @throws Exception
+     */
+    private function getProducerTopic(\RdKafka\Producer $producer, $topic): ProducerTopic
+    {
+        $pool = Snowflake::app()->getChannel();
+        return $pool->pop($topic . '::' . ProducerTopic::class, function () use ($producer, $topic) {
+            return $producer->newTopic($topic, $this->topic_conf);
+        });
+    }
 
 
-	/**
-	 * @param $message
-	 * @param null $key
-	 * @param bool $isAck
-	 * @throws NotFindClassException
-	 * @throws ReflectionException
-	 * @throws Exception
-	 */
-	public function delivery($message, $key = null, $isAck = false)
-	{
-		if (!$this->conf || !$this->topicConf) {
-			throw new Exception('Error. Please set kafka conf.');
-		}
-		$this->conf->setErrorCb(function ($kafka, $err, $reason) {
-			$this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
-		});
+    /**
+     * @param string $toPic
+     * @param string|null $key
+     * @param array $data
+     * @param string|null $groupId
+     */
+    public function batch(string $toPic, ?string $key, array $data, ?string $groupId = null)
+    {
+        $this->beforePushMessage($toPic, $groupId);
 
-		$event = Snowflake::app()->getEvent();
-		$event->on(Event::SYSTEM_RESOURCE_RELEASES, [$this, 'flush']);
-
-		if ($this->producer === null) {
-			$this->producer = Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]);
-		}
-
-		$this->setTopicAcks($isAck);
-		$this->push($message, $key);
-		if ($isAck === true) {
-			$this->flush();
-		}
-	}
+        $this->sendMessage($toPic, $data, $key);
+    }
 
 
-	/** @var ProducerTopic[] $topics */
-	private array $topics = [];
+    /**
+     * @param $topic
+     * @param $groupId
+     * @param $brokers
+     * @return ProducerTopic
+     * @throws \Snowflake\Exception\ConfigException
+     */
+    private function beforePushMessage($topic, $groupId): void
+    {
+        $consumers = Config::get('kafka.producers.' . $topic);
+        if (empty($consumers) || !is_array($consumers)) {
+            throw new Exception('You need set kafka.producers config');
+        }
+        if (!isset($consumers['brokers'])) {
+            throw new Exception('You need set brokers config.');
+        }
+        if (!empty($groupId)) {
+            $consumers['groupId'] = $groupId;
+        } else if (!isset($consumers['groupId'])) {
+            $consumers['groupId'] = $topic . ':' . Snowflake::localhost();
+        }
+        $this->setGroupId($consumers['groupId']);
+        $this->setBrokers($consumers['brokers']);
+        $this->setTopic($topic);
+    }
 
 
-	/**
-	 * @param $message
-	 * @param $key
-	 */
-	private function push($message, $key)
-	{
-		if (!isset($this->topics[$this->_topic])) {
-			$this->topics[$this->_topic] = $this->producer->newTopic($this->_topic, $this->topicConf);
-		}
-		$this->topics[$this->_topic]->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
-		$this->producer->poll(0);
-	}
+    /**
+     * @param TopicConf $topicConf
+     * @param string $topic
+     * @param array $message
+     * @param string $key
+     * @throws Exception
+     */
+    private function sendMessage(string $topic, array $message, string $key = '')
+    {
+        $producer = $this->getProducer();
+        $producerTopic = $this->getProducerTopic($producer, $topic);
+
+        if ($this->isAck) {
+            $this->flush($producer);
+        }
+
+        foreach ($message as $value) {
+            $producerTopic->produce(RD_KAFKA_PARTITION_UA, 0, $value, $key);
+            $producer->poll(0);
+        }
+        $this->flush($producer);
+        $this->recover($topic, $producer, $producerTopic);
+    }
 
 
-	/**
-	 * @param $isAck
-	 */
-	private function setTopicAcks(bool $isAck)
-	{
-		if ($isAck) {
-			if ($this->producer->getOutQLen() > 0) {
-				$this->flush();
-			}
-			$this->topicConf->set('request.required.acks', '1');
-		} else {
-			$this->topicConf->set('request.required.acks', '0');
-		}
-	}
+    /**
+     * @param bool $ack
+     */
+    public function setAsk(bool $ack){
+        $this->isAck = $ack;
+        $this->topicConf->set('request.required.acks', $this->isAck ? '1' : '0');
+    }
 
 
-	public function flush()
-	{
-		while ($this->producer->getOutQLen() > 0) {
-			$result = $this->producer->flush(100);
-			if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
-				break;
-			}
-		}
-	}
+
+    /**
+     * @param \RdKafka\Producer $producer
+     * @param ProducerTopic $producerTopic
+     * @throws Exception
+     */
+    private function recover(string $topic, \RdKafka\Producer $producer, ProducerTopic $producerTopic)
+    {
+        $channel = Snowflake::app()->getChannel();
+        $channel->push($producerTopic, $topic . '::' . ProducerTopic::class);
+        $channel->push($producer, \RdKafka\Producer::class);
+    }
+
+
+    /**
+     * @param \RdKafka\Producer $producer
+     */
+    public function flush(\RdKafka\Producer $producer)
+    {
+        while ($producer->getOutQLen() > 0) {
+            $result = $producer->flush(100);
+            if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
+                break;
+            }
+        }
+    }
 
 }
diff --git a/function.php b/function.php
index cefd3b59..fc68bb2c 100644
--- a/function.php
+++ b/function.php
@@ -588,6 +588,26 @@ if (!function_exists('listen')) {
 }
 
 
+if (!function_exists('event')) {
+
+
+    /**
+     * @param $name
+     * @param $callback
+     * @param $params
+     * @param $isAppend
+     * @throws Exception
+     * @throws Exception
+     */
+    function event($name, $callback, $params = [], $isAppend = true)
+    {
+        $event = Snowflake::app()->getEvent();
+        $event->on($name, $callback, $params, $isAppend);
+    }
+
+}
+
+
 if (!function_exists('alias')) {
 
     /**