diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 50ab91b..97b579f 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -2,93 +2,38 @@ name: PHPUnit
 
 on: [ push, pull_request ]
 
-env:
-  SWOOLE_VERSION: '4.8.6'
-  SWOW_VERSION: 'develop'
-
 jobs:
-  ci:
-    name: Test PHP ${{ matrix.php-version }} on ${{ matrix.engine }}
-    runs-on: "${{ matrix.os }}"
+  ci32:
+    runs-on: ubuntu-latest
+    container: hyperf/hyperf:${{ matrix.hyperf-version }}
     strategy:
       matrix:
-        os: [ ubuntu-latest ]
-        php-version: [ '8.0', '8.1' ]
-        engine: [ 'none', 'swoole', 'swow' ]
-      max-parallel: 6
+        hyperf-version:
+          - "8.2-alpine-v3.22-swoole"
+          - "8.3-alpine-v3.23-swoole"
+          - "8.4-alpine-v3.23-swoole"
       fail-fast: false
+      max-parallel: 15
+
     steps:
       - name: Checkout
-        uses: actions/checkout@v3
-      - name: Setup PHP
-        uses: shivammathur/setup-php@v2
-        with:
-          php-version: ${{ matrix.php-version }}
-          tools: phpize
-          ini-values: ｜
-            opcache.enable_cli=1
-
-          coverage: none
-      - name: Setup Swoole
-        if: ${{ matrix.engine == 'swoole' }}
+        uses: actions/checkout@v4
+      - name: Setup Environment
         run: |
-          sudo apt-get clean
-          sudo apt-get update
-          sudo apt-get upgrade -f
-          sudo apt-get install libcurl4-openssl-dev
-          wget https://github.com/swoole/swoole-src/archive/v${SWOOLE_VERSION}.tar.gz -O swoole.tar.gz
-          mkdir -p swoole
-          tar -xf swoole.tar.gz -C swoole --strip-components=1
-          rm swoole.tar.gz
-          cd swoole
-          phpize
-          ./configure --enable-openssl --enable-http2 --enable-swoole-curl --enable-swoole-json
-          make -j$(nproc)
-          sudo make install
-          sudo sh -c "echo extension=swoole > /etc/php/${{ matrix.php-version }}/cli/conf.d/swoole.ini"
+          pwd
+          ls -al
+          php -v
+          php -m
           php --ri swoole
-      - name: Setup Swow
-        if: ${{ matrix.engine == 'swow' }}
-        run: |
-          wget https://github.com/swow/swow/archive/"${SWOW_VERSION}".tar.gz -O swow.tar.gz
-          mkdir -p swow
-          tar -xf swow.tar.gz -C swow --strip-components=1
-          rm swow.tar.gz
-          cd swow/ext || exit
+          composer -V
 
-          phpize
-          ./configure --enable-debug
-          make -j "$(nproc)"
-          sudo make install
-          sudo sh -c "echo extension=swow > /etc/php/${{ matrix.php-version }}/cli/conf.d/swow.ini"
-          php --ri swow
-
-      - name: Setup Packages
+      - name: Install Dependencies
         run: |
-          composer update -o --no-scripts
-          composer update doctrine/instantiator:^1.0
-          composer require symfony/finder:^5.0
-          composer require hyperf/di
-          composer require hyperf/grpc-client
-          composer require hyperf/framework
-          composer require psr/log:^1.0
-          composer require psr/container:^1.0
-          composer require symfony/console:^5.0
-          composer require hyperf/dispatcher
-          composer require hyperf/server
-          composer require hyperf/http-server
-          composer require hyperf/db
-          composer require hyperf/db-connection
-          composer require hyperf/json-rpc
-          composer require hyperf/rpc-client
-          composer require symfony/serializer:^5.0
+          composer update -o 
+          composer info
 
-      - name: Setup Swow Packages
-        if: ${{ matrix.engine == 'swow' }}
-        run: |
-          composer require hyperf/engine-swow
+      - name: Static Analysis
+        run: composer analyse
 
-      - name: Run Test Cases
-        run: |
-          composer analyse
-          composer test
+      - name: Run Tests
+        run: composer test
\ No newline at end of file
diff --git a/README-CN.md b/README-CN.md
index b2a1c92..36a3688 100644
--- a/README-CN.md
+++ b/README-CN.md
@@ -73,6 +73,7 @@ composer require dtm/dtm-client
 ```bash
 php bin/hyperf.php vendor:publish dtm/dtm-client
 ```
+另需运行vendor/dtm/dtm-client/publish下的barrier表迁移文件
 
 如果您是在非 Hyperf 框架中使用，可复制 `./vendor/dtm/dtm-client/publish/dtm.php` 文件到对应的配置目录中。
 
@@ -374,3 +375,73 @@ class XAController
 
 ```
 上面的代码首先注册了一个全局XA事务，然后添加了两个子事务transIn、transOut。子事务全部执行成功之后，提交给dtm。dtm收到提交的xa全局事务后，会调用所有子事务的xa commit，完成整个xa事务。
+
+## 二阶段消息
+
+### 代码示例
+
+以下展示在 Hyperf 框架中的使用方法，其它框架类似
+
+```php
+<?php
+
+declare(strict_types=1);
+
+namespace App\Controller;
+
+use DtmClient\Barrier;
+use DtmClient\Middleware\DtmMiddleware;
+use DtmClient\Msg;
+use DtmClient\TransContext;
+use Hyperf\HttpServer\Annotation\Controller;
+use Hyperf\HttpServer\Annotation\RequestMapping;
+
+#[Controller(prefix: '/msg')]
+class MsgController extends AbstractSagaController
+{
+    private Msg $msg;
+
+    public function __construct(Msg $msg)
+    {
+        $this->msg = $msg;
+    }
+
+    #[RequestMapping(path: 'msg')]
+    public function msg()
+    {
+        $this->msg->init();
+        $this->msg->add('http://127.0.0.1:19501/msg/test', ['name' => 'dtmMsg']);
+        //添加Topic
+        //$this->msg->addTopic('TransIn', ['name' => 'Topic dtmMsg']);
+
+        $this->msg->doAndSubmitDB('http://127.0.0.1:19501/msg/queryPrepared', function () {
+            var_dump('执行业务');
+        });
+        return TransContext::getGid();
+    }
+
+    #[RequestMapping(path: 'queryPrepared')]
+    public function queryPrepared(Barrier $barrier)
+    {
+        var_dump(__METHOD__);
+        $transType = $this->request->query('trans_type');
+        $gid = $this->request->query('gid');
+        $barrier->queryPrepared($transType, $gid);
+    }
+
+    #[RequestMapping(path: 'test')]
+    public function test()
+    {
+        var_dump(__METHOD__);
+        return 'rest';
+    }
+
+    #[RequestMapping(path: 'subscribe')]
+    public function subscribe()
+    {
+        //订阅
+        $this->msg->subscribe('TransIn', 'http://127.0.0.1:19501/msg/test', 'subscribe test');
+        return 'subscribe';
+    }
+}
+```
\ No newline at end of file
diff --git a/composer.json b/composer.json
index d3a6f63..6eac07e 100644
--- a/composer.json
+++ b/composer.json
@@ -20,26 +20,35 @@
         }
     },
     "require": {
-        "php": ">=8.0",
+        "php": ">=8.2",
         "ext-json": "*",
         "guzzlehttp/guzzle": "^7.4",
-        "hyperf/context": "^2.2|^3.0|^3.1",
+        "hyperf/context": "~3.2.0",
         "psr/http-server-middleware": "^1.0"
     },
     "require-dev": {
+        "hyperf/config": "~3.2.0",
+        "hyperf/db": "~3.2.0",
+        "hyperf/redis": "~3.2.0",
+        "hyperf/di": "~3.2.0",
+        "hyperf/grpc-client": "~3.2.0",
+        "hyperf/rpc-client": "~3.2.0",
+        "hyperf/json-rpc": "~3.2.0",
+        "hyperf/http-server": "~3.2.0",
+        "hyperf/db-connection": "~3.2.0",
         "friendsofphp/php-cs-fixer": "^3.0",
         "mockery/mockery": "^1.0",
         "phpstan/phpstan": "^1.0",
         "phpunit/phpunit": "^9.5"
     },
     "suggest": {
-        "hyperf/config": "^2.2|^3.0|^3.1",
-        "hyperf/db": "^2,2|^3.0|^3.1",
-        "hyperf/redis": "^2,2|^3.0|^3.1",
-        "hyperf/di": "^2.2|^3.0|^3.1",
-        "hyperf/grpc-client": "^2.2|^3.0|^3.1",
-        "hyperf/rpc-client": "^2.2|^3.0|^3.1",
-        "hyperf/json-rpc": "^2.2|^3.0|^3.1",
+        "hyperf/config": "~3.2.0",
+        "hyperf/db": "~3.2.0",
+        "hyperf/redis": "~3.2.0",
+        "hyperf/di": "~3.2.0",
+        "hyperf/grpc-client": "~3.2.0",
+        "hyperf/rpc-client": "~3.2.0",
+        "hyperf/json-rpc": "~3.2.0",
         "ext-openssl": "Required to use HTTPS.",
         "ext-pdo": "Required to use MySQL Client.",
         "ext-pdo_mysql": "Required to use MySQL Client.",
diff --git a/phpstan.neon b/phpstan.neon
index 7e0bc12..dc005b9 100644
--- a/phpstan.neon
+++ b/phpstan.neon
@@ -5,16 +5,13 @@
 #
 parameters:
   level: 1
+  excludePaths:
+    - */src/DbTransaction/LaravelDbTransaction.php
   ignoreErrors:
-        - '#Call to static method beginTransaction\(\) on an unknown class Illuminate\\Support\\Facades\\DB\.#'
-        - '#Call to static method commit\(\) on an unknown class Illuminate\\Support\\Facades\\DB\.#'
-        - '#Call to static method rollback\(\) on an unknown class Illuminate\\Support\\Facades\\DB\.#'
         - '#Static call to instance method Hyperf\\DB\\DB::execute\(\)\.#'
         - '#Static call to instance method Hyperf\\DB\\DB::beginTransaction\(\)\.#'
         - '#Static call to instance method Hyperf\\DB\\DB::commit\(\)\.#'
         - '#Static call to instance method Hyperf\\DB\\DB::rollback\(\)\.#'
         - '#Call to static method currentRouteAction\(\) on an unknown class Illuminate\\Support\\Facades\\Route\.#'
         - '#Parameter \$request of method DtmClient\\Middleware\\DtmLaravelMiddleware::handle\(\) has invalid type Illuminate\\Http\\Request\.#'
-        - '#Call to static method affectingStatement\(\) on an unknown class Illuminate\\Support\\Facades\\DB\.#'
         - '#Static call to instance method Hyperf\\DB\\DB::query\(\)\.#'
-        - '#Call to static method select\(\) on an unknown class Illuminate\\Support\\Facades\\DB\.#'
diff --git a/publish/dtm.php b/publish/dtm.php
index 2c37ff7..2d6259b 100644
--- a/publish/dtm.php
+++ b/publish/dtm.php
@@ -19,6 +19,7 @@
     'barrier' => [
         'db' => [
             'type' => DbType::MySQL,
+            // 'type' => DbType::DB,
         ],
         'apply' => [],
     ],
diff --git a/src/Api/ApiInterface.php b/src/Api/ApiInterface.php
index dcd9084..4578b83 100644
--- a/src/Api/ApiInterface.php
+++ b/src/Api/ApiInterface.php
@@ -26,5 +26,7 @@ public function query(array $body);
 
     public function queryAll(array $body);
 
+    public function subscribe(array $body);
+
     public function transRequestBranch(RequestBranch $requestBranch);
 }
diff --git a/src/Api/GrpcApi.php b/src/Api/GrpcApi.php
index 5010457..546dffb 100644
--- a/src/Api/GrpcApi.php
+++ b/src/Api/GrpcApi.php
@@ -79,6 +79,11 @@ public function queryAll(array $body)
         throw new UnsupportedException('Unsupported QueryAll operation');
     }
 
+    public function subscribe(array $body)
+    {
+        throw new UnsupportedException('Unsupported QueryAll operation');
+    }
+
     public function transRequestBranch(RequestBranch $requestBranch)
     {
         [$hostname, $method] = $this->parseHostnameAndMethod($requestBranch->url);
diff --git a/src/Api/HttpApi.php b/src/Api/HttpApi.php
index 4377016..2c14b95 100644
--- a/src/Api/HttpApi.php
+++ b/src/Api/HttpApi.php
@@ -85,6 +85,11 @@ public function queryAll(array $body)
         return $this->transQuery($body, Operation::QUERY_ALL);
     }
 
+    public function subscribe(array $body)
+    {
+        return $this->transQuery($body, Operation::SUBSCRIBE);
+    }
+
     public function getClient(): Client
     {
         return $this->client;
diff --git a/src/Api/JsonRpcHttpApi.php b/src/Api/JsonRpcHttpApi.php
index a2f4817..d9ca45c 100644
--- a/src/Api/JsonRpcHttpApi.php
+++ b/src/Api/JsonRpcHttpApi.php
@@ -81,6 +81,11 @@ public function queryAll(array $body)
         throw new UnsupportedException('Unsupported Query operation');
     }
 
+    public function subscribe(array $body)
+    {
+        throw new UnsupportedException('Unsupported Query operation');
+    }
+
     public function getClient(): Client
     {
         return $this->client;
diff --git a/src/Barrier.php b/src/Barrier.php
index aea87e4..cc906bc 100644
--- a/src/Barrier.php
+++ b/src/Barrier.php
@@ -21,10 +21,13 @@ class Barrier
 
     protected MySqlBarrier $mySqlBarrier;
 
-    public function __construct(ConfigInterface $config, MySqlBarrier $mySqlBarrier)
+    protected DbBarrier $dbBarrier;
+
+    public function __construct(ConfigInterface $config, MySqlBarrier $mySqlBarrier, DbBarrier $dbBarrier)
     {
         $this->config = $config;
         $this->mySqlBarrier = $mySqlBarrier;
+        $this->dbBarrier = $dbBarrier;
     }
 
     public function call(callable $businessCall)
@@ -77,6 +80,8 @@ protected function getBarrier(): BarrierInterface
         switch ($this->config->get('dtm.barrier.db.type', DbType::MySQL)) {
             case DbType::MySQL:
                 return $this->mySqlBarrier;
+            case DbType::DB:
+                return $this->dbBarrier;
             default:
                 throw new UnsupportedException('Barrier DB type is unsupported.');
         }
diff --git a/src/Constants/DbType.php b/src/Constants/DbType.php
index 8fabd53..780e42d 100644
--- a/src/Constants/DbType.php
+++ b/src/Constants/DbType.php
@@ -12,5 +12,7 @@ class DbType
 {
     public const MySQL = 'mysql';
 
+    public const DB = 'db';
+
     public const Redis = 'redis';
 }
diff --git a/src/Constants/Operation.php b/src/Constants/Operation.php
index 053fa41..c7d6c73 100644
--- a/src/Constants/Operation.php
+++ b/src/Constants/Operation.php
@@ -31,4 +31,6 @@ class Operation
     public const ACTION = 'action';
 
     public const ROLLBACK = 'rollback';
+
+    public const SUBSCRIBE = 'subscribe';
 }
diff --git a/src/DbBarrier.php b/src/DbBarrier.php
new file mode 100644
index 0000000..29038a1
--- /dev/null
+++ b/src/DbBarrier.php
@@ -0,0 +1,61 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of DTM-PHP.
+ *
+ * @license  https://github.com/dtm-php/dtm-client/blob/master/LICENSE
+ */
+namespace DtmClient;
+
+use DtmClient\Constants\Branch;
+use DtmClient\Constants\Operation;
+use DtmClient\Exception\FailureException;
+
+class DbBarrier extends MySqlBarrier
+{
+    public function queryPrepared(string $transType, string $gid): bool
+    {
+        $db = $this->DBTransaction->connection();
+        $table = $db->raw('barrier');
+        $db->table($table)->insertOrIgnore([
+            'trans_type' => $transType,
+            'gid' => $gid,
+            'branch_id' => Branch::MsgDoBranch0,
+            'op' => Branch::MsgDoOp,
+            'barrier_id' => Branch::MsgDoBarrier1,
+            'reason' => Operation::ROLLBACK,
+        ]);
+
+        $reason = $db->table($table)
+            ->select('reason')
+            ->where([
+                'gid' => $gid,
+                'branch_id' => Branch::MsgDoBranch0,
+                'op' => Branch::MsgDoOp,
+                'barrier_id' => Branch::MsgDoBarrier1,
+            ])->first();
+
+        if ($reason->reason == Operation::ROLLBACK) {
+            throw new FailureException();
+        }
+        return true;
+    }
+
+    protected function insertBarrier(string $transType, string $gid, string $branchId, string $op, string $barrierID, string $reason): int
+    {
+        if (empty($op)) {
+            return 0;
+        }
+        $db = $this->DBTransaction->connection();
+        $table = $db->raw('barrier');
+        return $db->table($table)->insertOrIgnore([
+            'trans_type' => $transType,
+            'gid' => $gid,
+            'branch_id' => $branchId,
+            'op' => $op,
+            'barrier_id' => $barrierID,
+            'reason' => $reason,
+        ]);
+    }
+}
diff --git a/src/DbTransaction/DBTransactionInterface.php b/src/DbTransaction/DBTransactionInterface.php
index 8915b84..2222ab2 100644
--- a/src/DbTransaction/DBTransactionInterface.php
+++ b/src/DbTransaction/DBTransactionInterface.php
@@ -22,6 +22,11 @@ public function execute(string $sql, array $bindings = []): int;
 
     public function query(string $sql, array $bindings = []): bool|array;
 
+    /**
+     * @return \Hyperf\Database\ConnectionInterface|\Illuminate\Database\ConnectionInterface
+     */
+    public function connection();
+
     public function xaExecute(string $sql, array $bindings = []): int;
 
     public function xaQuery(string $sql, array $bindings = []): bool|array;
diff --git a/src/DbTransaction/HyperfDbTransaction.php b/src/DbTransaction/HyperfDbTransaction.php
index efbc1cb..d99a64b 100644
--- a/src/DbTransaction/HyperfDbTransaction.php
+++ b/src/DbTransaction/HyperfDbTransaction.php
@@ -42,4 +42,9 @@ public function query(string $sql, array $bindings = []): bool|array
     {
         return Db::select($sql, $bindings);
     }
+
+    public function connection()
+    {
+        return Db::connection();
+    }
 }
diff --git a/src/DbTransaction/HyperfSimpleDbTransaction.php b/src/DbTransaction/HyperfSimpleDbTransaction.php
index 9dd8ea7..ab47f8a 100644
--- a/src/DbTransaction/HyperfSimpleDbTransaction.php
+++ b/src/DbTransaction/HyperfSimpleDbTransaction.php
@@ -8,6 +8,7 @@
  */
 namespace DtmClient\DbTransaction;
 
+use Exception;
 use Hyperf\Contract\ConfigInterface;
 use Hyperf\DB\DB;
 
@@ -42,5 +43,9 @@ public function query(string $sql, array $bindings = []): bool|array
     {
         return DB::query($sql, $bindings);
     }
-    
+
+    public function connection()
+    {
+        throw new Exception('Not implemented');
+    }
 }
diff --git a/src/DbTransaction/LaravelDbTransaction.php b/src/DbTransaction/LaravelDbTransaction.php
index b98a480..3838fab 100644
--- a/src/DbTransaction/LaravelDbTransaction.php
+++ b/src/DbTransaction/LaravelDbTransaction.php
@@ -1,11 +1,7 @@
 <?php
 
 declare(strict_types=1);
-/**
- * This file is part of DTM-PHP.
- *
- * @license  https://github.com/dtm-php/dtm-client/blob/master/LICENSE
- */
+
 namespace DtmClient\DbTransaction;
 
 use Illuminate\Support\Facades\DB;
@@ -39,4 +35,9 @@ public function query(string $sql, array $bindings = []): bool|array
     {
         return DB::select($sql, $bindings);
     }
+
+    public function connection()
+    {
+        return DB::connection();
+    }
 }
diff --git a/src/JsonRpc/JsonRpcClient.php b/src/JsonRpc/JsonRpcClient.php
index 15ae095..e118f05 100644
--- a/src/JsonRpc/JsonRpcClient.php
+++ b/src/JsonRpc/JsonRpcClient.php
@@ -17,6 +17,8 @@
 use Hyperf\RpcClient\Exception\RequestException;
 use Psr\Container\ContainerInterface;
 
+use function Hyperf\Support\make;
+
 class JsonRpcClient extends AbstractServiceClient
 {
     public function __construct(ContainerInterface $container)
diff --git a/src/JsonRpc/JsonRpcClientManager.php b/src/JsonRpc/JsonRpcClientManager.php
index a5ee657..db491c2 100644
--- a/src/JsonRpc/JsonRpcClientManager.php
+++ b/src/JsonRpc/JsonRpcClientManager.php
@@ -8,6 +8,8 @@
  */
 namespace DtmClient\JsonRpc;
 
+use function Hyperf\Support\make;
+
 class JsonRpcClientManager
 {
     /**
diff --git a/src/Middleware/DtmMiddleware.php b/src/Middleware/DtmMiddleware.php
index 9c785d3..33eed7b 100644
--- a/src/Middleware/DtmMiddleware.php
+++ b/src/Middleware/DtmMiddleware.php
@@ -16,6 +16,7 @@
 use DtmClient\Exception\OngingException;
 use DtmClient\Exception\RuntimeException;
 use Hyperf\Contract\ConfigInterface;
+use Hyperf\Contract\StdoutLoggerInterface;
 use Hyperf\Di\Annotation\AnnotationCollector;
 use Hyperf\Grpc\StatusCode;
 use Hyperf\HttpMessage\Stream\SwooleStream;
@@ -33,11 +34,14 @@ class DtmMiddleware implements MiddlewareInterface
 
     protected ConfigInterface $config;
 
-    public function __construct(Barrier $barrier, ResponseInterface $response, ConfigInterface $config)
+    protected StdoutLoggerInterface $logger;
+
+    public function __construct(Barrier $barrier, ResponseInterface $response, ConfigInterface $config, StdoutLoggerInterface $logger)
     {
         $this->barrier = $barrier;
         $this->response = $response;
         $this->config = $config;
+        $this->logger = $logger;
     }
 
     public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface
@@ -137,6 +141,7 @@ protected function handlerBarrierCall(callable $businessCall): ResponseInterface
             $this->isGRPC() && $response = $response->withTrailer('grpc-status', (string) $failureException->getCode())->withTrailer('grpc-message', $failureException->getMessage());
             return $response;
         } catch (\Throwable $throwable) {
+            $this->logger->error((string)$throwable);
             $code = $this->isGRPC() ? 200 : Result::FAILURE_STATUS;
             $response = $response->withStatus($code);
             $this->isGRPC() && $response = $response->withTrailer('grpc-status', (string) Result::FAILURE_STATUS)->withTrailer('grpc-message', $throwable->getMessage());
diff --git a/src/Msg.php b/src/Msg.php
index 71d2d71..c41efe8 100644
--- a/src/Msg.php
+++ b/src/Msg.php
@@ -16,7 +16,6 @@
 use DtmClient\Exception\UnsupportedException;
 use Google\Protobuf\GPBEmpty;
 use Google\Protobuf\Internal\Message;
-use GuzzleHttp\Psr7\Response;
 
 class Msg extends AbstractTransaction
 {
@@ -36,6 +35,12 @@ public function init(?string $gid = null)
         TransContext::init($gid, TransType::MSG, '');
     }
 
+    public function addTopic(string $action, $payload)
+    {
+        $action = sprintf('topic://%s', $action);
+        $this->add($action, $payload);
+    }
+
     public function add(string $action, $payload)
     {
         TransContext::addStep(['action' => $action]);
@@ -64,6 +69,11 @@ public function submit()
         return $this->api->submit(TransContext::toArray());
     }
 
+    public function doAndSubmitDB(string $queryPrepared, callable $businessCall)
+    {
+        $this->doAndSubmit($queryPrepared, fn () => $this->barrier->call($businessCall));
+    }
+
     public function doAndSubmit(string $queryPrepared, callable $businessCall)
     {
         $this->barrier->barrierFrom(TransType::MSG, TransContext::getGid(), '00', 'msg');
@@ -83,6 +93,14 @@ public function doAndSubmit(string $queryPrepared, callable $businessCall)
         }
     }
 
+    public function subscribe(string $topic, string $url, string $remark = '')
+    {
+        $body['topic'] = $topic;
+        $body['url'] = $url;
+        $body['remark'] = $remark;
+        return $this->api->subscribe($body);
+    }
+
     protected function queryPrepared(string $queryPrepared)
     {
         // If busicall return an error other than failure, we will query the result
diff --git a/src/MySqlBarrier.php b/src/MySqlBarrier.php
index d30c23d..d581d5d 100644
--- a/src/MySqlBarrier.php
+++ b/src/MySqlBarrier.php
@@ -15,6 +15,7 @@
 use DtmClient\DbTransaction\DBTransactionInterface;
 use DtmClient\Exception\DuplicatedException;
 use DtmClient\Exception\FailureException;
+use Psr\Http\Message\ResponseInterface;
 
 class MySqlBarrier implements BarrierInterface
 {
@@ -40,6 +41,7 @@ public function call(callable $businessCall): bool
         $originOP = [
             Branch::BranchCancel => Branch::BranchTry,
             Branch::BranchCompensate => Branch::BranchAction,
+            Branch::BranchRollback => Branch::BranchAction,
         ][$op] ?? '';
 
         $this->DBTransaction->beginTransaction();
@@ -53,7 +55,7 @@ public function call(callable $businessCall): bool
             }
 
             if (
-                ($op == Operation::BRANCH_CANCEL || $op == Operation::BRANCH_COMPENSATE) && $originAffected > 0 // null compensate
+                ($op == Operation::BRANCH_CANCEL || $op == Operation::BRANCH_COMPENSATE || $op == Operation::ROLLBACK) && $originAffected > 0 // null compensate
                 || $currentAffected == 0// repeated request or dangled request
             ) {
                 $this->DBTransaction->commit();
@@ -61,7 +63,7 @@ public function call(callable $businessCall): bool
             }
 
             $response = $businessCall();
-            if ($response->getStatusCode() !== Result::SUCCESS_STATUS) {
+            if ($response instanceof ResponseInterface && $response->getStatusCode() !== Result::SUCCESS_STATUS) {
                 throw new FailureException();
             }
 
diff --git a/src/Saga.php b/src/Saga.php
index d3acdd2..9d51bfc 100644
--- a/src/Saga.php
+++ b/src/Saga.php
@@ -12,14 +12,15 @@
 use DtmClient\Api\ApiInterface;
 use DtmClient\Constants\Protocol;
 use DtmClient\Constants\TransType;
+use DtmClient\Context\Context;
 use DtmClient\Exception\UnsupportedException;
 use Google\Protobuf\Internal\Message;
 
 class Saga extends AbstractTransaction
 {
-    protected array $orders = [];
+    protected const CONCURRENT = self::class . '.concurrent';
 
-    protected bool $concurrent = false;
+    protected const ORDERS = self::class . '.orders';
 
     protected ApiInterface $api;
 
@@ -33,6 +34,8 @@ public function init(?string $gid = null)
         if ($gid === null) {
             $gid = $this->generateGid();
         }
+        Context::set(static::CONCURRENT, false);
+        Context::set(static::ORDERS, []);
         TransContext::init($gid, TransType::SAGA, '');
     }
 
@@ -60,13 +63,15 @@ public function add(string $action, string $compensate, array|object $payload):
 
     public function addBranchOrder(int $branch, array $preBranches): static
     {
-        $this->orders[$branch] = $preBranches;
+        $orders = Context::get(static::ORDERS, []);
+        $orders[$branch] = $preBranches;
+        Context::set(static::ORDERS, $orders);
         return $this;
     }
 
     public function enableConcurrent()
     {
-        $this->concurrent = true;
+        Context::set(static::CONCURRENT, true);
     }
 
     public function submit()
@@ -77,10 +82,10 @@ public function submit()
 
     public function addConcurrentContext()
     {
-        if ($this->concurrent) {
+        if (Context::get(static::CONCURRENT, false)) {
             TransContext::setCustomData(json_encode([
-                'concurrent' => $this->concurrent,
-                'orders' => $this->orders ?: null,
+                'concurrent' => Context::get(static::CONCURRENT),
+                'orders' => Context::get(static::ORDERS) ?: null,
             ]));
         }
     }
diff --git a/src/TCC.php b/src/TCC.php
index ec313dd..8ccc447 100644
--- a/src/TCC.php
+++ b/src/TCC.php
@@ -16,6 +16,7 @@
 use DtmClient\Exception\InvalidArgumentException;
 use DtmClient\Exception\UnsupportedException;
 use Google\Protobuf\Internal\Message;
+use Psr\Http\Message\ResponseInterface;
 
 class TCC extends AbstractTransaction
 {
@@ -51,6 +52,7 @@ public function globalTransaction(callable $callback, ?string $gid = null)
 
     /**
      * @param array|Message $body
+     * @return array|ResponseInterface
      */
     public function callBranch($body, string $tryUrl, string $confirmUrl, string $cancelUrl)
     {
@@ -73,6 +75,7 @@ public function callBranch($body, string $tryUrl, string $confirmUrl, string $ca
                 $branchRequest->branchId = $branchId;
                 $branchRequest->op = Operation::TRY;
                 $branchRequest->body = $body;
+                $branchRequest->branchHeaders = TransContext::getBranchHeaders();
                 return $this->api->transRequestBranch($branchRequest);
             case Protocol::GRPC:
                 if (! $body instanceof Message) {
diff --git a/tests/Cases/Api/GrpcApiTest.php b/tests/Cases/Api/GrpcApiTest.php
index 9d5b587..1171876 100644
--- a/tests/Cases/Api/GrpcApiTest.php
+++ b/tests/Cases/Api/GrpcApiTest.php
@@ -11,10 +11,10 @@
 use DtmClient\Api\GrpcApi;
 use DtmClient\Grpc\GrpcClientManager;
 use DtmClientTest\Cases\AbstractTestCase;
+use Hyperf\Context\ApplicationContext;
+use Hyperf\Coroutine\Channel\Pool as ChannelPool;
 use Hyperf\Contract\ConfigInterface;
 use Hyperf\GrpcClient\BaseClient;
-use Hyperf\Utils\ApplicationContext;
-use Hyperf\Utils\ChannelPool;
 use Mockery;
 use Psr\Container\ContainerInterface;
 
diff --git a/tests/Cases/BarrierTest.php b/tests/Cases/BarrierTest.php
index 26c54d2..cde9637 100644
--- a/tests/Cases/BarrierTest.php
+++ b/tests/Cases/BarrierTest.php
@@ -6,6 +6,7 @@
 use DtmClient\Constants\DbType;
 use DtmClient\Constants\Protocol;
 use DtmClient\Constants\TransType;
+use DtmClient\DbBarrier;
 use DtmClient\MySqlBarrier;
 use DtmClient\TransContext;
 use Hyperf\Contract\ConfigInterface;
@@ -21,7 +22,11 @@ public function testCall()
 
         $mySqlBarrier->method('call')->willReturn(true);
 
-        $barrier = new Barrier($configInterface, $mySqlBarrier);
+        $dbBarrier = $this->createMock(DbBarrier::class);
+
+        $dbBarrier->method('call')->willReturn(true);
+
+        $barrier = new Barrier($configInterface, $mySqlBarrier, $dbBarrier);
         $this->assertTrue($barrier->call(function () {
             return true;
         }));
@@ -34,7 +39,9 @@ public function testBarrierFrom()
 
         $mySqlBarrier = $this->createMock(MySqlBarrier::class);
 
-        $barrier = new Barrier($configInterface, $mySqlBarrier);
+        $dbBarrier = $this->createMock(DbBarrier::class);
+
+        $barrier = new Barrier($configInterface, $mySqlBarrier, $dbBarrier);
         $barrier->barrierFrom(TransType::TCC, 'gid', 'branchId', 'try', 'phase2Url', 'testDtm');
 
         $this->assertSame(TransContext::toArray(), [
diff --git a/tests/Cases/SagaTest.php b/tests/Cases/SagaTest.php
index c2b7b94..93d2cb6 100644
--- a/tests/Cases/SagaTest.php
+++ b/tests/Cases/SagaTest.php
@@ -5,12 +5,21 @@
 use DtmClient\Api\ApiInterface;
 use DtmClient\Constants\Protocol;
 use DtmClient\Constants\TransType;
+use DtmClient\Context\Context;
 use DtmClient\Grpc\Message\DtmBranchRequest;
 use DtmClient\Saga;
 use DtmClient\TransContext;
 
 class SagaTest extends AbstractTestCase
 {
+    protected function setUp(): void
+    {
+        parent::setUp();
+        // 清理 Context，避免测试间状态泄漏
+        Context::set('DtmClient\Saga.concurrent', false);
+        Context::set('DtmClient\Saga.orders', []);
+    }
+
     public function testInit()
     {
         $api = \Mockery::mock(ApiInterface::class);
@@ -83,9 +92,7 @@ public function testAddBranchOrder()
         $saga->addBranchOrder(1, ['preBranches']);
         $saga->addBranchOrder(2, ['preBranches1']);
 
-        $ordersProperty = new \ReflectionProperty($saga, 'orders');
-        $ordersProperty->setAccessible(true);
-        $orders = $ordersProperty->getValue($saga);
+        $orders = Context::get('DtmClient\Saga.orders');
         $this->assertEquals([1 => ['preBranches'], 2 => ['preBranches1']], $orders);
     }
 
@@ -97,9 +104,7 @@ public function testEnableConcurrent()
 
         $saga->enableConcurrent();
 
-        $concurrentProperty = new \ReflectionProperty($saga, 'concurrent');
-        $concurrentProperty->setAccessible(true);
-        $concurrent = $concurrentProperty->getValue($saga);
+        $concurrent = Context::get('DtmClient\Saga.concurrent');
         $this->assertTrue($concurrent);
     }
 
@@ -110,10 +115,8 @@ public function testEnableConcurrentWithoutAddBranchOrders()
         $saga = new Saga($api);
         $saga->enableConcurrent();
 
-        $ordersProperty = new \ReflectionProperty($saga, 'orders');
-        $ordersProperty->setAccessible(true);
-        $orders = $ordersProperty->getValue($saga);
-        $this->assertSame(null, $orders ?: null);
+        $orders = Context::get('DtmClient\Saga.orders');
+        $this->assertSame([], $orders);
     }
 
     public function testSubmit()