add: fix

application/modules/Pay/controllers/Kafka.php

 <?php
 
-use Api\PhpServices\Idgen\Idgen;
 use App\Base\Base;
-
-use App\Models\order\mysql\PayOrderItem;
-use App\Services\pay\PayService;
-use App\Services\refund\RefundService;
-use App\Models\order\mysql\CallbackLog;
-use Helpers\DingTalk;
+use Api\PhpUtils\Kafka\KafkaUtil;
 
 class TccController extends Base
 {
-    /**
-     * tcc test
-     */
-    public function tryAction()
-    {
-        if(!empty($this->params['sleep'])) {
-            sleep($this->params['sleep']);
-        }
-
-        if(!empty($this->params['error'])) {
-            throw new \App\Exception\BaseException(['msg'=>'try is error', 'code'=>2001]);
-        }
-
-        $this->success(['result'=>[]]);
-    }
-
-    public function mainAction() {
-        if(!empty($this->params['sleep'])) {
-            sleep($this->params['sleep']);
-        }
-
-        if(!empty($this->params['error'])) {
-            $this->success(['data'=>2]);
-        }
-
-        $this->success(['data'=>1]);
-    }
-
-    public function sleepAction() {
-
-        if(!empty($this->params['sleep'])) {
-            sleep($this->params['sleep']);
-        }
-
-        $this->success(['result'=>$this->params]);
-
-    }
-
-    public function confirmAction() {
-
-        if(!empty($this->params['sleep'])) {
-            sleep($this->params['sleep']);
-        }
-
-        if(!empty($this->params['error'])) {
-            throw new \App\Exception\BaseException(['msg'=>'confirm is error', 'code'=>2001]);
-        }
-
-        $this->success(['result'=>[]]);
-    }
-
-    public function cancelAction() {
-
-        if(!empty($this->params['sleep'])) {
-            sleep($this->params['sleep']);
-        }
-
-        if(!empty($this->params['error'])) {
-            throw new \App\Exception\BaseException(['msg'=>'confirm is error', 'code'=>2001]);
-        }
-
-        $this->success(['result'=>[]]);
-    }
-
-    public function traceAction() {
-        $params = $this->getRequest()->getPost();
-        //$params['traceId'] = \Api\PhpUtils\Log\Tracer::getTraceId();
+    public function productAction() {
+
+        $topicName = 'bp_pay_callback.broker_list';
+
+
+        $setConfDrMsgCb = function ($kafka, \RdKafka\Message $message) {
+            echo '打印消息:' . "\n";
+            var_export($message);
+            echo "\n";
+            if($message->err){
+                //@todo 生产失败的逻辑处理
+            } else{
+                //@todo 生产成功的逻辑处理
+            }
+        };
+        $setErrorCbCallback = function ($kafka, $err, $reason) {
+            echo sprintf("setErrorCb (error_ori:%s)(error: %s) (reason: %s)", $err, rd_kafka_err2str($err), $reason);
+        };
+        // $kafka = KafkaUtil::getInstance('test', ['context' => ['contextArr']]);
+        // $ret = $kafka->produce('eee');
+
+
+        $brokers = config("kafka", "bp_pay_sync_account.broker_list");
+        $topic = config("kafka", "bp_pay_sync_account.topic");
+        $kafka = KafkaUtil::getInstance();
+        $ret = $kafka->setBrokers($brokers)->setTopicName('topicA')->produce('ccc');
+        var_dump($ret);
+        var_dump($kafka);
 
-        file_put_contents('/tmp/aa.log', json_encode($_SERVER), FILE_APPEND);
-        $params['traceId'] = \Helpers\Tracer::getTraceId();
-        $this->success($params);
     }
 
-    public function idsAction() {
-        $number = $this->params['number'] ?? '12';
-        $count = $this->params['count'] ?? 5;
-        $res = Idgen::get(appConfig('idgen.partner'), appConfig('idgen.key'), [], [[
-            "type" => "55c768",
-            'number' => $number,
-            "count" => $count]]);
-        $id = $res['id_datetime']['55c768'] ?? [];
+    public function consumeAction() {
 
-        $this->success(['ids'=>$id]);
     }
 }

application/services/pay/PayService.php

@@ -743,6 +743,7 @@ class PayService
     /**
      * @param $payer
      * @throws PayException
+     * @throws BaseException
      * 负责支付的清分
      */
     private function make_order_clearing($payer)
@@ -1100,8 +1101,7 @@ class PayService
                         'remark' => '',
                     ];
 
-                } else {
-                }
+                } else {}
 
                 $this->clear_items_list[] = [
                     'pay_order_clearing_item_id' => array_shift($this->gen_ids),