制作一个text协议的worker来消费新增节点的任务

EdgeManager.php

@@ -1,5 +1,6 @@
 <?php
 use Workerman\Worker;
+use \Workerman\Connection\AsyncTcpConnection;
 use Workerman\Connection\TcpConnection;
 use Workerman\Protocols\Http\Request;
 use Workerman\Protocols\Http\Response;
@@ -12,14 +13,16 @@ $options = getopt('h::', ['server_name:', 'user:', 'password:', 'help::']);
 init_db($options['server_name'], $options['user'], $options['password']);
 
 $worker = new Worker('http://0.0.0.0:8888');
-$worker -> name = 'CaptureWorker';
+$worker -> name = 'EntryPoint';
 $worker -> count = 20;
 
 $worker -> onWorkerStart = function(Worker $worker) {
     global $options, $dbconn;
 
-    $dbconn = pg_connect(sprintf(  "host=%s dbname=scada user=%s password=%s",
+    $dbconn = pg_connect(sprintf(
-                                    $options['server_name'], $options['user'], $options['password']));
+        "host=%s dbname=scada user=%s password=%s",
+        $options['server_name'], $options['user'], $options['password']
+    ));
 };
 
 $worker -> onMessage = function(TcpConnection $connection, Request $request) {
@@ -44,9 +47,9 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
                 $response = new Response(200, [
                     'Content-Type' => 'application/json;charset=utf-8',
                 ], json_encode(array(
-                    'action' => '错得太离谱以至于我也不知道这是什么动作',
+                    'action'    => '错得太离谱以至于我也不知道这是什么动作',
-                    'errcode' => 4001,
+                    'errcode'   => 4001,
-                    'errmsg' => 'POST的内容不是JSON，也并非可按照JSON解析的字符串！或者检查一下你的JSON是不是有注释？'
+                    'errmsg'    => 'POST的内容不是JSON，也并非可按照JSON解析的字符串！或者检查一下你的JSON是不是有注释？'
                 )));
                 $connection -> send($response);
             }
@@ -61,23 +64,35 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
                 // 因为pg_insert()不需要action字段
                 $action = $post -> action;
                 unset($post -> action);
-                $enode_configure = new ENodeConfigure($dbconn, post: $post);
+
-                $res = $enode_configure -> $action();
+                if ($action === 'add_node') {
-                if ($res === true)
+                    $task_connection = new AsyncTcpConnection('Text://127.0.0.1:1888');
-                    $connection -> send(json_encode(array(
+                    $task_data = array(
-                        'code'  => 0,
+                        'action' => 'add_node',
-                        'msg'   => 'Success'
+                        'data'   => $post,
-                    )));
+                    );
-                else if ($res === "REPLICATED")
+                    $task_connection -> send(json_encode($task_data));
-                    $connection -> send(json_encode(array(
+                    $task_connection -> onMessage = function(AsyncTcpConnection $task_connection, $task_result) use ($connection) {
-                        'code'  => 1,
+                        $task_connection -> close();
-                        'msg'   => '同一工序单元内的节点编码不可重复！'
+                        $connection -> send($task_result);
-                    )));
+                    };
-                else if ($res === false) {
+                    // 执行异步连接
-                    $connection -> send(json_encode(array(
+                    $task_connection->connect();
-                        'code'  => 1,
+                } else {
-                        'msg'   => '服务器内部逻辑错误，请联系开发者！'
+                    $enode_configure = new ENodeConfigure($dbconn, post: $post);
-                    )));
+                    $res = $enode_configure -> $action();
+                    
+                    if ($res === true)
+                        $connection -> send(json_encode(array(
+                            'code'  => 0,
+                            'msg'   => 'Success'
+                        )));
+                    else if ($res === false) {
+                        $connection -> send(json_encode(array(
+                            'code'  => 1,
+                            'msg'   => '服务器内部逻辑错误，请联系开发者！'
+                        )));
+                    }
                 }                
             } else if ($post -> action === 'set_node_data') {
                 $data_capture = new EDataCapture($dbconn, post: $post);
@@ -206,4 +221,44 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
     }
 };
 
+$consumer = new Worker('Text://0.0.0.0:1888');
+$consumer -> name = 'Consumer';
+
+$consumer -> onWorkerStart = function(Worker $consumer) {
+    global $options, $consumer_dbconn;
+
+    $consumer_dbconn = pg_connect(sprintf(
+        "host=%s dbname=scada user=%s password=%s",
+        $options['server_name'], $options['user'], $options['password']
+    ));
+};
+
+$consumer -> onMessage = function(TcpConnection $connection, $task_data) {
+    global $consumer_dbconn;
+
+    $task_data = json_decode($task_data);
+
+    if ($task_data -> action === 'add_node') {
+        $enode_configure = new ENodeConfigure($consumer_dbconn, post: $task_data -> data);
+        $res = $enode_configure -> add_node();
+    
+        if ($res === true)
+            $connection -> send(json_encode(array(
+                'code'  => 0,
+                'msg'   => 'Success'
+            )));
+        else if ($res === "REPLICATED")
+            $connection -> send(json_encode(array(
+                'code'  => 1,
+                'msg'   => '同一工序单元内的节点编码不可重复！'
+            )));
+        else if ($res === false) {
+            $connection -> send(json_encode(array(
+                'code'  => 1,
+                'msg'   => '服务器内部逻辑错误，请联系开发者！'
+            )));
+        }
+    }
+};
+
 Worker::runAll();

EdgeManager/EDataCapture/ENodeConfigure.php

@@ -9,6 +9,7 @@ class ENodeConfigure {
     ) {}
 
     function add_node() {
+        pg_query($this -> dbconn, "BEGIN");
         $exists = pg_query($this -> dbconn, sprintf(
             "SELECT EXISTS(
                 SELECT 1 FROM hf_mes_scada_data_capture_node_configure
@@ -16,12 +17,13 @@ class ENodeConfigure {
                 AND working_subclass = '%s'
             )", $this -> post -> code, $this -> post -> working_subclass
         ));
-        if (pg_fetch_assoc($exists)['exists'] == 't') {
+        if (pg_fetch_assoc($exists)['exists'] === 't') {
+            pg_query($this -> dbconn, "ROLLBACK");
             return "REPLICATED";
         }
 
         $table_name = 'hf_mes_scada_data_capture_node_data_' . $this -> post -> working_subclass;
-        pg_query($this -> dbconn, "BEGIN");
+        // https://www.postgresql.org/docs/current/when-can-parallel-query-be-used.html
         $res[] = pg_query($this -> dbconn, sprintf(
             'CREATE TABLE IF NOT EXISTS "%s" (
                 id serial8,