使用强大的DBPack处理分布式事务（PHP使用教程）

07 00 00 // 前 3 个字节表示 payload 的长度为 7 个字节

01 // sequence 响应的序号，前 4 个字节一起构成了 OKPacket 的 heade

00 // 标识 payload 为 OKPacket

00 // affected row

00 // last insert id

03 00 // 状态标志位

00 00 // warning 数量

ETCD\_VER=v3.5.3



# choose either URL

GOOGLE\_URL=https://storage.googleapis.com/etcd

GITHUB\_URL=https://github.com/etcd-io/etcd/releases/download

DOWNLOAD\_URL=${GOOGLE\_URL}



rm -f /tmp/etcd-${ETCD\_VER}-linux-amd64.tar.gz

rm -rf /tmp/etcd-download-test && mkdir -p /tmp/etcd-download-test



curl -L ${DOWNLOAD\_URL}/${ETCD\_VER}/etcd-${ETCD\_VER}-linux-amd64.tar.gz -o /tmp/etcd-${ETCD\_VER}-linux-amd64.tar.gz

tar xzvf /tmp/etcd-${ETCD\_VER}-linux-amd64.tar.gz -C /tmp/etcd-download-test --strip-components=1

rm -f /tmp/etcd-${ETCD\_VER}-linux-amd64.tar.gz



/tmp/etcd-download-test/etcd --version

/tmp/etcd-download-test/etcdctl version

/tmp/etcd-download-test/etcdutl version

-- ----------------------------

-- Table structure for undo\_log

-- ----------------------------

DROP TABLE IF EXISTS `undo\_log`;

CREATE TABLE `undo\_log` (

  `id` bigint(20) NOT NULL AUTO\_INCREMENT,

  `branch\_id` bigint(20) NOT NULL,

  `xid` varchar(100) NOT NULL,

  `context` varchar(128) NOT NULL,

  `rollback\_info` longblob NOT NULL,

  `log\_status` int(11) NOT NULL,

  `log\_created` datetime NOT NULL,

  `log\_modified` datetime NOT NULL,

  `ext` varchar(100) DEFAULT NULL,

  PRIMARY KEY (`id`),

  KEY `idx\_unionkey` (`xid`,`branch\_id`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

# 更新distributed\_transaction.etcd\_config.endpoints

# 更新listeners配置项，调整为实际聚合层服务的地址和端口

# 更新filters配置项，配置聚合层服务的API endpoint

vim /path/to/your/aggregation-service/config-aggregation.yaml



# 更新distributed\_transaction.etcd\_config.endpoints

# 更新listeners配置项，配置业务数据库信息，包括dbpack代理的端口

# 更新data\_source\_cluster.dsn

vim /path/to/your/business-service/config-service.yaml

git clone git@github.com:cectc/dbpack.git



cd dbpack

# build on local env

make build-local

# build on production env

make build



./dist/dbpack start --config /path/to/your/config-aggregation.yaml



./dist/dbpack start --config /path/to/your/config-service.yaml

server {

    listen 3001; # 暴露的服务端口

    index index.php index.html;

    root /var/www/code/; # 业务代码根目录



    location / {

        try\_files $uri /index.php?$args;

    }



    location ~ \.php$ {

        fastcgi\_split\_path\_info ^(.+\.php)(/.+)$;

        fastcgi\_pass order-svc-app:9000; # php-fpm 端口

        fastcgi\_index index.php;

        include fastcgi\_params;

        fastcgi\_param SCRIPT\_FILENAME $document\_root$fastcgi\_script\_name;

        fastcgi\_param PATH\_INFO $fastcgi\_path\_info;

    }

}

class AggregationSvc

{



    public function CreateSo(string $xid, bool $rollback): bool

    {

        $createSoSuccess = $this->createSoRequest($xid);

        if (!$createSoSuccess) {

            return false;

        }

        $allocateInventorySuccess = $this->allocateInventoryRequest($xid);

        if (!$allocateInventorySuccess) {

            return false;

        }

        if ($rollback) {

            return false;

        }

        return true;

    }



    // private function createSoRequest(string $xid) ...

    // private function allocateInventoryRequest(string $xid) ...

}



$reqPath = strtok($\_SERVER["REQUEST\_URI"], '?');

$reaHeaders = getallheaders();



$xid = $reaHeaders['X-Dbpack-Xid'] ?? '';



if (empty($xid)) {

    die('xid is not provided!');

}



$aggregationSvc = new AggregationSvc();



if ($\_SERVER['REQUEST\_METHOD'] === 'POST') {

    switch ($reqPath) {

        case '/v1/order/create':

            if ($aggregationSvc->CreateOrder($xid, false)) {

                responseOK();

            } else {

                responseError();

            }

        case '/v1/order/create2':

            if ($aggregationSvc->CreateSo($xid, true)) {

                responseOK();

            } else {

                responseError();

            }

            break;

        default:

            die('api not found');

    }

}



function responseOK() {

    http\_response\_code(200);

    echo json\_encode([

        'success' => true,

        'message' => 'success',

    ]);

}



function responseError() {

    http\_response\_code(400);

    echo json\_encode([

        'success' => false,

        'message' => 'fail',

    ]);

}

class OrderDB

{

    private PDO $\_connection;

    private static OrderDB $\_instance;

    private string $\_host = 'dbpack-order';

    private int $\_port = 13308;

    private string $\_username = 'dksl';

    private string $\_password = '123456';

    private string $\_database = 'order';



    const insertSoMaster = "INSERT /\*+ XID('%s') \*/ INTO order.so\_master (sysno, so\_id, buyer\_user\_sysno, seller\_company\_code, 

        receive\_division\_sysno, receive\_address, receive\_zip, receive\_contact, receive\_contact\_phone, stock\_sysno, 

        payment\_type, so\_amt, status, order\_date, appid, memo) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,now(),?,?)";



    const insertSoItem = "INSERT /\*+ XID('%s') \*/ INTO order.so\_item(sysno, so\_sysno, product\_sysno, product\_name, cost\_price, 

        original\_price, deal\_price, quantity) VALUES (?,?,?,?,?,?,?,?)";



    public static function getInstance(): OrderDB

    {

        if (empty(self::$\_instance)) {

            self::$\_instance = new self();

        }

        return self::$\_instance;

    }



    private function \_\_construct()

    {

        try {

            $this->\_connection = new PDO(

                "mysql:host=$this->\_host;port=$this->\_port;dbname=$this->\_database;charset=utf8",

                $this->\_username,

                $this->\_password,

                [

                    PDO::ATTR\_PERSISTENT => true,

                    PDO::ATTR\_EMULATE\_PREPARES => false, // to let DBPack handle prepread sql

                ]

            );

        } catch (PDOException $e) {

            die($e->getMessage());

        }

    }



    private function \_\_clone()

    {

    }



    public function getConnection(): PDO

    {

        return $this->\_connection;

    }



    public function createSo(string $xid, array $soMasters): bool

    {

        $this->getConnection()->setAttribute(PDO::ATTR\_ERRMODE, PDO::ERRMODE\_EXCEPTION);

        try {

            $this->getConnection()->beginTransaction();

            foreach ($soMasters as $master) {

                if (!$this->insertSo($xid, $master)) {

                    throw new PDOException("failed to insert soMaster");

                }

            }

            $this->getConnection()->commit();

        } catch (PDOException $e) {

            $this->getConnection()->rollBack();

            return false;

        }

        return true;

    }



    private function insertSo(string $xid, array $soMaster): bool

    {

        // insert into so\_master, so\_item ...

    }

}



$reqPath = strtok($\_SERVER["REQUEST\_URI"], '?');

$reqHeaders = getallheaders();



$xid = $reqHeaders['Xid'] ?? '';



if (empty($xid)) {

    die('xid is not provided!');

}



if ($\_SERVER['REQUEST\_METHOD'] === 'POST') {

    if ($reqPath === '/createSo') {

        $reqBody = file\_get\_contents('php://input');

        $soMasters = json\_decode($reqBody, true);



        $orderDB = OrderDB::getInstance();

        $result = $orderDB->createSo($xid, $soMasters);



        if ($result) {

            responseOK();

        } else {

            responseError();

        }

    }

}



function responseOK() {

    http\_response\_code(200);

    echo json\_encode([

        'success' => true,

        'message' => 'success',

    ]);

}



function responseError() {

    http\_response\_code(400);

    echo json\_encode([

        'success' => false,

        'message' => 'fail',

    ]);

}

curl -X{HTTP Method} http://localhost:{DBPack监听的聚合层服务端口}/{聚合层服务的API endpoint}