现状

  1. 没有原生PHP pulsar extension
  2. pulsar上开启了 KOP（Kafka On Pulsar）或 WebSocket 协议
  3. 客户端使用 rdkafka的扩展或 websocket 链接

实际使用情况

  • saas-project 使用 kop。实际是 rdkafka 扩展。生产者发送消息时 kop 协议会对消息体做一定的处理。（base64 encode）
  • his-api 使用 websocket方式。生产者发送消息时 ws 协议会对消息体做一定的处理。（base64 encode）。使用 textalk/websocket 作为管理 websocket 生命周期的依赖。新增了一个 Utils 类在项目中。
  1. <?php
  4. namespace App\Utils;
  6. use Illuminate\Support\Carbon;
  7. use Illuminate\Support\Facades\Log;
  8. use WebSocket\Client;
  10. /**
  11.  * pulsar web socket
  12.  * Class PulsarUtils
  13.  * @package App\Utils
  14.  * @resource https://pulsar.apache.org/docs/en/client-libraries-websocket/
  15.  * @example
  16.  *  PulsarUtils::getInstance()->setConf([
  17.  * "url" => "111.111.0.60:8080",
  18.  * "tenant"=>"socket",
  19.  * "namespace"=>"socket_namespace",
  20.  * "topic"=>"socket_topic",
  21.  * "subname"=>"ctz_test",
  22.  * ])->consumerMsg(function ($msgReceive){
  23.  * echo base64_decode($msgReceive['payload']).PHP_EOL;
  24.  * return true;
  25.  * });
  26.  *
  27.  */
  28. final class PulsarUtils
  29. {
  30.     protected $pulsarConf = [];
  32.     private static $_instance = NULL;
  34.     public static function getInstance()
  35.     {
  36.         if (is_null(self::$_instance)) {
  37.             self::$_instance = new self();
  38.         }
  39.         return self::$_instance;
  40.     }
  42.     public function checkConfig($configArray = [])
  43.     {
  44.         if (!array_key_exists("url", $configArray)) {
  45.             return false;
  46.         }
  48.         if (!array_key_exists("tenant", $configArray)) {
  49.             return false;
  50.         }
  52.         if (!array_key_exists("namespace", $configArray)) {
  53.             return false;
  54.         }
  57.         if (!array_key_exists("topic", $configArray)) {
  58.             return false;
  59.         }
  61.         return true;
  62.     }
  64.     /**
  65.      * @param $pulsarConfig pulsar配置
  66.      * @return $this
  67.      */
  68.     public function setConf($pulsarConfig)
  69.     {
  70.         if ($this->checkConfig($pulsarConfig)) {
  71.             $this->pulsarConf = $pulsarConfig;
  72.         }
  73.         return $this;
  74.     }
  76.     /**
  77.      * 获取配置
  78.      * @return array
  79.      */
  80.     public function getConf()
  81.     {
  82.         return $this->pulsarConf;
  83.     }
  85.     /**
  86.      * 生产者 pulsar ws链接
  87.      * @param $config
  88.      * @return string
  89.      */
  90.     protected function getProducerWsUrl()
  91.     {
  92.         if ($this->checkConfig($this->pulsarConf)) {
  93.             return sprintf(
  94.                 "ws://%s/ws/v2/producer/persistent/%s/%s/%s",
  95.                 $this->pulsarConf['url'], $this->pulsarConf['tenant'], $this->pulsarConf['namespace'], $this->pulsarConf['topic']
  96.             );
  97.         }
  98.         return "";
  99.     }
  101.     /**
  102.      * 消费者 pulsar ws链接
  103.      * @return string
  104.      */
  105.     protected function getConsumerWSUrl()
  106.     {
  107.         if ($this->checkConfig($this->pulsarConf)) {
  108.             return sprintf(
  109.                 "ws://%s/ws/v2/consumer/persistent/%s/%s/%s/%s?subscriptionType=Shared",
  110.                 $this->pulsarConf['url'], $this->pulsarConf['tenant'], $this->pulsarConf['namespace'], $this->pulsarConf['topic'], $this->pulsarConf['subname'] ?? ""
  111.             );
  112.         }
  113.         return "";
  114.     }
  116.     /**
  117.      * 生产者端
  118.      * @param $config
  119.      * @return null|Client
  120.      */
  121.     protected function producerClient()
  122.     {
  123.         $wsUrl = $this->getProducerWsUrl($this->pulsarConf);
  124.         static $pulsarProducerClients = [];
  125.         if ($wsUrl) {
  126.             $wsUrlHash = md5($wsUrl);
  127.             if (isset($pulsarProducerClients[$wsUrlHash])) {
  128.                 return $pulsarProducerClients[$wsUrlHash];
  129.             } else {
  130.                 $pulsarProducerClients[$wsUrlHash] = new Client($wsUrl);
  131.                 return $pulsarProducerClients[$wsUrlHash];
  132.             }
  133.         }
  134.         return null;
  135.     }
  137.     /**
  138.      * @param $config
  139.      * @return Client
  140.      * @throws \Exception
  141.      */
  142.     protected function consumerClient()
  143.     {
  144.         $wsUrl = $this->getConsumerWSUrl($this->pulsarConf);
  145.         static $pulsarConsumerClients = [];
  146.         if ($wsUrl) {
  147.             $wsUrlHash = md5($wsUrl);
  148.             if (isset($pulsarConsumerClients[$wsUrlHash])) {
  149.                 return $pulsarConsumerClients[$wsUrlHash];
  150.             } else {
  151.                 $pulsarConsumerClients[$wsUrlHash] = new Client($wsUrl,["timeout"=>10]);
  152.                 return $pulsarConsumerClients[$wsUrlHash];
  153.             }
  154.         }
  155.         return null;
  156.     }
  158.     /**
  159.      * 生产者
  160.      * @param $message
  161.      * @return bool
  162.      * @throws \Exception
  163.      */
  164.     public function producerMsg($message)
  165.     {
  166.         $producerClient = $this->producerClient();
  167.         if ($producerClient == null) {
  168.             throw new \Exception("pulsar_producerMsg_exception:" . json_encode($this->getConf()));
  169.         }
  170.         //判断是否是数组类型
  171.         if (is_array($message)) {
  172.             $message = json_encode($message, JSON_UNESCAPED_UNICODE);
  173.         }
  175.         //发送前需要base64
  176.         $base64Message = base64_encode($message);
  178.         $prepareSendMessage = [
  179.             "payload"    => $base64Message,
  180.             "properties" => new \stdClass(),
  181.             "context"    => "",
  182.             "key"        => "",
  183.         ];
  185.         $sendMessage = json_encode($prepareSendMessage);
  186.         $producerClient->text($sendMessage);
  187.         $sendResult = $producerClient->receive();
  188.         try {
  189.             $result = json_decode($sendResult, true);
  190.             if (array_key_exists("result", $result) &&
  191.                 $result['result'] == "ok") {
  192.                 return true;
  194.             } else {
  195.                 throw new \Exception("pulsar_websocket_send:" .
  196.                     $sendResult);
  197.             }
  198.         } catch (\Exception $e) {
  199.             throw new \Exception("pulsar_send_exception:" . $e->getMessage());
  200.         }
  202.     }
  204.     /**
  205.      * 消费者
  206.      * @param callable $func
  207.      * @throws \Exception
  208.      */
  209.     public function consumerMsg(callable $func)
  210.     {
  211.         $consumerClient = $this->consumerClient();
  212.         if ($consumerClient == null) {
  213.             throw new \Exception("pulsar_consumerMsg_exception:" . json_encode($this->getConf()));
  214.         }
  215.         while (true) {
  216.             try {
  217.                 $result = $consumerClient->receive();
  218.                 if($result==null){
  219.                     continue;
  220.                 }
  221.                 Log::channel('pulsar-his')->info("pulsar_receive",["receive_time"=>Carbon::now()->toDateTimeString(),"receive_data"=>$result]);
  222.                 $resultArray = json_decode($result, true);
  223.                 $functionResult = call_user_func($func, $resultArray);
  224.                 if ($functionResult) {
  225.                     $ackMsg = ['messageId' => $resultArray['messageId']];
  226.                     $consumerClient->text(json_encode($ackMsg));
  227.                 }
  228.             } catch (\Throwable $e) {
  229.                 //连接超时 默认5S连接超时
  230.                 if($e->getCode()==1024){
  231.                     //echo $e->getMessage().PHP_EOL;
  232.                     continue;
  233.                 }
  234.                 Log::channel('pulsar-his')->error("解析消费异常:",[$e->getMessage(),$e->getCode(),$e->getLine()]);
  235.                 exit;
  236.             }
  237.         }
  238.     }
  239. }

存在问题

  1. 没有统一的 composer php 包依赖
  2. 封装比较粗糙
  3. 没有使用文档及 examples
  4. 错误处理较弱
  5. 稳定和可靠性待实际验证

最佳实践 RoadMap

  1. 开发自有 pulsar 的 websocket 通讯方式的 PHP composer依赖包。
  2. composer包要实现主要的 pulsar API 。包括但不限于客户端链接、断开、生产者发送消息、消费者读取消费消息、消费者回执等。保证客户端的稳定，不能阻塞后续程序执行，不能有内存泄漏，不能导致程序非正常退出，要正确处理pulsar server端的错误。
  3. 依赖包最好适配框架 Laravel.
  4. Topic 命名要能区分
    1. 业务方向或业务分组
    2. 以小写字母加数字以下划线_和.命名
  5. 消息体大小尽量不要超过 1MB
  6. 确认消费完成后，消费者需要回执 Ack
  7. 对于重复消费行为，消费者需要针对业务做幂等或业务逻辑要求的处理
  8. 生产和消费可溯源，可追踪，有日志留存
  9. 针对消费失败和其他故障的可重放
  10. 管理平台或面板展示

监控 && 告警指标

  1. 各Topic producer 和 consumer 的连接情况包括
    1. 正在活跃
    2. 历史连接
    3. 消息积压
    4. 消费速度
  2. producer 和 consumer 无存活告警、队列消息积压条数条件告警
  3. pulsar 本身告警（pulsar提供）
    1. cpu
    2. 内存
    3. 心跳
    4. 存储