DTM

https://dtm.pub
Admin
DTM is an open-source distributed transaction manager that solves data consistency across databases,...
Tokens:6,677
Snippets:72
Trust Score:2.9
Update:4 months ago
Show doc for...
Tokens:
Raw
### Run DTM SAGA Example

Source: https://dtm.pub/guide/e-saga

Command to clone and run the DTM SAGA example application.

```bash
git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_saga_barrier
```

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

### Install DTM Helm Chart

Source: https://dtm.pub/deploy/deploy.html

Installs the DTM chart into a specified namespace. Ensure the namespace exists or use `--create-namespace`.

```bash
helm install --create-namespace -n dtm-system dtm ./charts

```

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

### Running the XA Transaction Example

Source: https://dtm.pub/practice/xa.html

Command to run the full XA transaction example using the dtm CLI. This command executes the main application and initiates the HTTP XA flow.

```bash
go run main.go http_xa
```

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

### Start DTM Binary

Source: https://dtm.pub/deploy/deploy.html

Start the DTM service using a compiled binary and a configuration file. DTM listens on HTTP port 36789 and gRPC port 36790.

```bash
./dtm -c ./conf.yml

```

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

### XA Transaction SQL Commands

Source: https://dtm.pub/practice/xa.html

Demonstrates the SQL commands for starting, preparing, and committing an XA transaction. Ensure all participants are ready before committing.

```sql
XA start '4fPqCNTYeSG' -- 开启一个 xa 事务
UPDATE `user_account` SET `balance`=balance + 30,`update_time`='2021-06-09 11:50:42.438' WHERE user_id = '1'
XA end '4fPqCNTYeSG'
XA prepare '4fPqCNTYeSG' -- 此调用之前，连接断开，那么事务会自动回滚
-- 当所有的参与者完成了prepare，就进入第二阶段 提交
xa commit '4fPqCNTYeSG'
```

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

### Run DTM Server

Source: https://dtm.pub/guide/e-saga

Command to clone and run the DTM server. This is a prerequisite for running SAGA examples.

```bash
git clone https://github.com/dtm-labs/dtm && cd dtm
go run main.go
```

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

### Initialize Workflow SDK with gRPC

Source: https://dtm.pub/practice/workflow

Initializes the Workflow SDK for gRPC communication. This setup is required before using Workflow features with gRPC services. It configures the DTM server address, business server address, and the gRPC server instance.

```go
import 	"github.com/dtm-labs/dtmgrpc/workflow"

// 初始化workflow SDK，三个参数分别为：
// 第一个参数，dtm服务器地址
// 第二个参数，业务服务器地址
// 第三个参数，grpcServer
// workflow的需要从"业务服务器地址"+"grpcServer"上接收dtm服务器的回调
workflow.InitGrpc(dtmGrpcServer, busi.BusiGrpc, gsvr)
```

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

### Simulate TransIn Failure

Source: https://dtm.pub/guide/e-saga

Example of modifying the TransIn endpoint to return dtmcli.ErrFailure, simulating a transaction failure.

```go
app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  return dtmcli.ErrFailure
}))
```

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

### Register Workflow Handler

Source: https://dtm.pub/practice/workflow

Registers a workflow handler function with a specific name. This function defines the transactional steps, including forward operations and rollback callbacks for each branch. The registration must occur after the business service starts.

```go
wfName := "wf_saga"
err := workflow.Register(wfName, func(wf *workflow.Workflow, data []byte) error {
  req := MustUnmarshalReqGrpc(data)
  wf.NewBranch().OnRollback(func(bb *dtmcli.BranchBarrier) error {
    _, err := busi.BusiCli.TransOutRevert(wf.Context, req)
    return err
  })
  _, err := busi.BusiCli.TransOut(wf.Context, req)
  if err != nil {
    return err
  }
  wf.NewBranch().OnRollback(func(bb *dtmcli.BranchBarrier) error {
    _, err := busi.BusiCli.TransInRevert(wf.Context, req)
    return err
  })
  _, err = busi.BusiCli.TransIn(wf.Context, req)
  return err
})
```

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

### Initiating an XA Global Transaction with dtm

Source: https://dtm.pub/practice/xa.html

Shows how to initiate a global XA transaction using dtm's Go SDK. This involves registering the transaction and calling branches within a callback function.

```go
gid := dtmcli.MustGenGid(dtmutil.DefaultHTTPServer)
err := dtmcli.XaGlobalTransaction(dtmutil.DefaultHTTPServer, gid, func(xa *dtmcli.Xa) (*resty.Response, error) {
	resp, err := xa.CallBranch(&busi.TransReq{Amount: 30}, busi.Busi+"/TransOutXa")
	if err != nil {
		return resp, err
	}
	return xa.CallBranch(&busi.TransReq{Amount: 30}, busi.Busi+"/TransInXa")
})
```

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

### Create User Account Table

Source: https://dtm.pub/guide/e-saga

SQL statement to create the user account table with balance and trading balance fields.

```sql
CREATE TABLE dtm_busi.`user_account` (
  `id` int(11) AUTO_INCREMENT PRIMARY KEY,
  `user_id` int(11) not NULL UNIQUE ,
  `balance` decimal(10,2) NOT NULL DEFAULT '0.00',
  `trading_balance` decimal(10,2) NOT NULL DEFAULT '0.00',
  `create_time` datetime DEFAULT now(),
  `update_time` datetime DEFAULT now()
);
```

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

### 秒杀接口核心代码

Source: https://dtm.pub/app/flash.html

秒杀接口的核心代码，使用dtm框架进行二阶段消息处理，实现库存扣减和订单创建的原子性。库存扣减通过Redis实现，支持高并发。

```go
gid := "{a}flash-sale-" + activityID + "-" + uid
msg := dtmcli.NewMsg(DtmServer, gid). 
	Add(busi.Busi+"/createOrder", gin.H{activity_id: activityID, UID: uid})
err := msg.DoAndSubmit(busi.Busi+"/QueryPreparedRedis", func(bb *BranchBarrier) error {
  return bb.RedisCheckAdjustAmount(rds, "{a}stock-"+stockID, -1, 86400)
})
```

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

### Initialize Workflow SDK with HTTP

Source: https://dtm.pub/practice/workflow

Initializes the Workflow SDK for HTTP communication with the DTM server. This allows DTM SDK to interact with the DTM server using HTTP interfaces.

```go
workflow.InitHTTP(dtmHTTPClient)
```

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

### Compile DTM Binary

Source: https://dtm.pub/deploy/deploy.html

Compile the DTM binary from source code. Requires Go 1.16 or higher.

```bash
go build

```

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

### Run DTM with Docker

Source: https://dtm.pub/deploy/deploy.html

Launch DTM in a Docker container, exposing its HTTP and gRPC ports and configuring the storage backend via environment variables.

```bash
docker run --name dtm -p 36789:36789 -p 36790:36790 -e STORE_DRIVER=mysql -e STORE_HOST=localhost -e STORE_USER=root -e STORE_PASSWORD= -e STORE_PORT=3306 yedf/dtm:latest

```

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

### 反查接口核心代码

Source: https://dtm.pub/app/flash.html

秒杀接口的反查核心代码，用于dtm框架在事务中断时查询库存扣减是否成功。对于Redis数据，只需复制粘贴此代码即可。

```go
app.GET(BusiAPI+"/QueryPreparedRedis", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
		return MustBarrierFromGin(c).RedisQueryPrepared(rds)
	}))
```

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

### 运行测试

Source: https://dtm.pub/app/flash.html

执行秒杀性能测试的命令。

```shell
sh test-flash-sales.sh
```

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

### Kubernetes Deployment for DTM

Source: https://dtm.pub/deploy/deploy.html

A Kubernetes Deployment configuration for DTM, including resource requests, probes, and volume mounts for configuration.

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: dtm
  labels:
    app: dtm
spec:
  replicas: 2
  selector:
    matchLabels:
      app: dtm
  template:
    metadata:
      labels:
        app: dtm
    spec:
      containers:
        - name: dtm
          image: yedf/dtm:latest
          imagePullPolicy: IfNotPresent
          args:
            - "-c=/app/dtm/configs/config.yaml"
          volumeMounts:
            - mountPath: /app/dtm/configs
              name: config
          ports:
            - containerPort: 36789
              protocol: TCP
              name: http
            - containerPort: 36790
              protocol: TCP
              name: grpc
          livenessProbe:
            httpGet:
              path: /api/ping
              port: 36789
              scheme: HTTP
          readinessProbe:
            httpGet:
              path: /api/ping
              port: 36789
              scheme: HTTP
          resources:
            requests:
              cpu: 200m
              memory: 200Mi
      volumes:
        - name: config
          configMap:
            name: dtm-conf
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: dtm-conf
  labels:
    app: dtm
data:
  config.yaml: |-
    Store:
      Driver: mysql # 此处以 mysql 为例，其他数据库可自行替换
      Host: dtm-db  # 此处设置为集群外部的数据库 host，或者集群内部的数据库 svc-dns
      Port: 3306
      User: root
      Password: ''
---
apiVersion: v1
kind: Service
metadata:
  name: dtm-svc
  labels:
    app: dtm
spec:
  ports:
    - port: 36790
      targetPort: 36790
      name: grpc
      appProtocol: grpc # Kubernetes v1.20 [stable]，低版本请剔除此行
    - port: 36789
      targetPort: 36789
      name: http
      appProtocol: http # Kubernetes v1.20 [stable]，低版本请剔除此行
  selector:
    app: dtm
  type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: dtm-ing
spec:
  rules:
    - host: "your-domain.com"
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: dtm-svc
                port:
                  number: 36789 # 此处为 http server，grpc server 的设置，请访问 https://kubernetes.github.io/ingress-nginx/examples/grpc/
  ingressClassName: nginx # 使用了其他的 ingressClassName， 请自行查询

```

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

### 准备dtm

Source: https://dtm.pub/app/flash.html

在另一台阿里云主机上准备dtm的安装和配置步骤。包括安装git、golang，克隆dtm仓库并编译，以及修改dtm测试脚本以指向正确的Redis实例。

```shell
apt update
apt install -y git

wget https://golang.org/dl/go1.17.1.linux-amd64.tar.gz
rm -rf /usr/local/go && tar -C /usr/local -xzf go1.17.1.linux-amd64.tar.gz && cp -f /usr/local/go/bin/go /usr/local/bin/go

git clone https://github.com/dtm-labs/dtm.git && cd dtm && git checkout v1.11.0 && cd bench && make
# 修改 dtm/bench/test-flash-sales.sh
# export BUSI_REDIS=localhost:6379 => 虚拟机A 的私网ip

```

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

### 准备Redis

Source: https://dtm.pub/app/flash.html

在阿里云主机上准备Redis的安装和配置步骤。包括安装redis，修改配置文件以允许外部访问，并重启redis服务。

```shell
apt-get install -y redis
# 修改 /etc/redis/redis.conf
# bind 127.0.0.1 => 0.0.0.0

systemctl redis restart

```

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

### Initiate and Submit SAGA Transaction

Source: https://dtm.pub/guide/e-saga

Go code to create a new SAGA transaction, add branches with forward and compensating operations, and submit it for execution.

```go
	req := &gin.H{"amount": 30} // 微服务的载荷
	// DtmServer为DTM服务的地址
	saga := dtmcli.NewSaga(DtmServer, shortuuid.New()).
		// 添加一个TransOut的子事务，正向操作为url: qsBusi+"/TransOut"， 逆向操作为url: qsBusi+"/TransOutCom"
		Add(qsBusi+"/SagaBTransOut", qsBusi+"/SagaBTransOutCom", req).
		// 添加一个TransIn的子事务，正向操作为url: qsBusi+"/TransOut"， 逆向操作为url: qsBusi+"/TransInCom"
		Add(qsBusi+"/SagaBTransIn", qsBusi+"/SagaBTransInCom", req)
	// 提交saga事务，dtm会完成所有的子事务/回滚所有的子事务
	err := saga.Submit()
```

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

### Execute Workflow

Source: https://dtm.pub/practice/workflow

Executes a registered workflow by its name and provides initial data. The result indicates whether the global transaction succeeded (nil/ErrFailure) or requires further retries by DTM.

```go
req := &busi.ReqGrpc{Amount: 30}
err = workflow.Execute(wfName, shortuuid.New(), dtmgimp.MustProtoMarshal(req))
```

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

### Simulating XA Transaction Failure

Source: https://dtm.pub/practice/xa.html

Prepares a TransReq object with a specific result to trigger a failure in the XA transaction's second phase, leading to rollback.

```go
req := &busi.TransReq{Amount: 30, TransInResult: "FAILURE"}
```

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

### Perform Local Transaction Operation

Source: https://dtm.pub/practice/workflow

Demonstrates performing a local transaction operation within a workflow branch. This allows integrating native database transactions directly into the distributed workflow.

```go
wf.Do(func(bb *dtmcli.BranchBarrier) ([]byte, error) {
  return nil, bb.CallWithDB(dbGet(), func(tx *sql.Tx) error {
    return busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount, "")
  })
})
```

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

### 测试结果

Source: https://dtm.pub/app/flash.html

秒杀性能测试的输出结果，显示每秒完成的秒杀请求数。

```text
Requests per second:    11970.21 [#/sec] (mean)

```

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

### Handle TransIn Failure with Barrier

Source: https://dtm.pub/guide/e-saga

Modified TransIn endpoint demonstrating how BranchBarrier handles failures occurring before or after commit, ensuring correct compensating operations.

```go
app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, "")
  })
  return dtmcli.ErrFailure
}))
```

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

### Docker Compose for DTM

Source: https://dtm.pub/deploy/deploy.html

Defines a Docker Compose service for DTM, specifying the image, environment variables for storage configuration, and port mappings.

```yaml
version: '3'
services:
  dtm:
    image: yedf/dtm
    environment:
      STORE_DRIVER: mysql
      STORE_HOST: localhost
      STORE_USER: root
      STORE_PASSWORD: ''
      STORE_PORT: 3306
    ports:
      - '36789:36789'
      - '36790:36790'

```

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

### SAGA Transactional Endpoints

Source: https://dtm.pub/guide/e-saga

Gin framework endpoints for SAGA transaction branches (trans-in, trans-out) and their compensating operations. Utilizes BranchBarrier for idempotency.

```go
app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, "")
  })
}))
app.POST(BusiAPI+"/SagaBTransInCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransInUID, -reqFrom(c).Amount, "")
  })
}))
app.POST(BusiAPI+"/SagaBTransOut", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransOutUID, -reqFrom(c).Amount, "")
  })
}))
app.POST(BusiAPI+"/SagaBTransOutCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransOutUID, reqFrom(c).Amount, "")
  })
}))
```

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

### Saga Adjust Balance Function

Source: https://dtm.pub/guide/e-saga

Go function to adjust user account balance. Used for both forward and compensating operations in SAGA.

```go
func SagaAdjustBalance(db dtmcli.DB, uid int, amount int, result string) error {
	_, err := dtmimp.DBExec(db, "update dtm_busi.user_account set balance = balance + ? where user_id = ?", amount, uid)
	return err
}
```

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

### Uninstall DTM Helm Chart

Source: https://dtm.pub/deploy/deploy.html

Removes the DTM deployment and associated resources managed by the Helm chart.

```bash
helm delete -n dtm-system dtm

```

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

### Upgrade DTM Helm Chart

Source: https://dtm.pub/deploy/deploy.html

Upgrades an existing DTM deployment managed by Helm. This applies configuration changes or new chart versions.

```bash
helm upgrade -n dtm-system dtm ./charts

```

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

### Implementing XA Local Transaction for Branch Operations

Source: https://dtm.pub/practice/xa.html

Defines the handler for a branch operation within an XA transaction. It uses dtmcli.XaLocalTransaction to manage the local part of the XA transaction.

```go
app.POST(BusiAPI+"/TransInXa", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
	return dtmcli.XaLocalTransaction(c.Request.URL.Query(), BusiConf, func(db *sql.DB, xa *dtmcli.Xa) error {
		return AdjustBalance(db, TransInUID, reqFrom(c).Amount, reqFrom(c).TransInResult)
	})
}))
app.POST(BusiAPI+"/TransOutXa", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
	return dtmcli.XaLocalTransaction(c.Request.URL.Query(), BusiConf, func(db *sql.DB, xa *dtmcli.Xa) error {
		return AdjustBalance(db, TransOutUID, reqFrom(c).Amount, reqFrom(c).TransOutResult)
	})
}))
```

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

### Add Workflow Interceptor to gRPC Client

Source: https://dtm.pub/practice/workflow

Adds the workflow interceptor to a gRPC client. This interceptor automatically records RPC request results with DTM and retrieves saved results during retries, ensuring proper transaction state management.

```go
conn1, err := grpc.Dial(busi.BusiGrpc, grpc.WithUnaryInterceptor(workflow.Interceptor), nossl)
busi.BusiCli = busi.NewBusiClient(conn1)
```

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

### BranchBarrier Call Function

Source: https://dtm.pub/guide/e-saga

Signature for the Call function within BranchBarrier, used to ensure idempotency of sub-transaction operations.

```go
func (bb *BranchBarrier) Call(tx *sql.Tx, busiCall BarrierBusiFunc) error
```
Tokens:
Raw