docker run ok

Dockerfile

@@ -3,4 +3,4 @@ WORKDIR /app/dtm
 RUN go env -w GO111MODULE=on
 RUN go env -w GOPROXY=https://mirrors.aliyun.com/goproxy/,direct
 EXPOSE 8080
-CMD [ "/bin/bash", "-c", "go build app/main.go && /app/dtm/main dtmsvr" ]
+CMD [ "/bin/bash", "-c", "go build app/main.go && /app/dtm/main" ]

app/main.go

@@ -13,9 +13,10 @@ type M = map[string]interface{}
 
 func main() {
 	if len(os.Args) == 1 { // 默认情况下，展示saga例子
-		examples.PopulateMysql()
+		dtmsvr.PopulateMysql()
 		go dtmsvr.StartSvr()
 		go examples.SagaStartSvr()
+		time.Sleep(100 * time.Millisecond)
 		examples.SagaFireRequest()
 	} else if os.Args[1] == "dtmsvr" {
 		go dtmsvr.StartSvr()

common/conf.sample

@@ -1,6 +0,0 @@
-mysql:
-  host: 'xxx'
-  user: 'xxx'
-  password: 'xxx'
-  database: 'xxx'
-  port: '3306'

dtmsvr/conf.sample

@@ -1,6 +0,0 @@
-mysql:
-  host: 'xxx'
-  user: 'xxx'
-  password: 'xxx'
-  database: 'xxx'
-  port: '3306'

dtmsvr/dtmsvr.sql

@@ -2,53 +2,58 @@ CREATE DATABASE IF NOT EXISTS `dtm` /*!40100 DEFAULT CHARACTER SET utf8mb4 */;
 
 use dtm;
 
-drop table IF EXISTS saga;
-CREATE TABLE IF NOT EXISTS `saga` (
+drop table IF EXISTS trans_global;
+CREATE TABLE if not EXISTS `trans_global` (
   `id` int(11) NOT NULL AUTO_INCREMENT,
-  `gid` varchar(45) NOT NULL COMMENT '事务全局id',
-  `steps` json NOT NULL COMMENT 'saga的所有步骤',
-  `status` varchar(45) NOT NULL COMMENT '全局事务的状态 prepared | processing | succeed | failed',
-  `trans_query` varchar(128) NOT NULL COMMENT '事务未决状态的查询api',
-  `finish_time` datetime DEFAULT NULL,
-  `rollback_time` datetime DEFAULT NULL,
+  `gid` varchar(128) NOT NULL COMMENT '事务全局id',
+  `trans_type` varchar(45) not null COMMENT '事务类型: saga | xa',
+  `data` TEXT COMMENT '事务携带的数据',
+  `status` varchar(45) NOT NULL COMMENT '全局事务的状态 prepared | committed | finished | rollbacked',
+  `query_prepared` varchar(128) NOT NULL COMMENT 'prepared状态事务的查询api',
   `create_time` datetime DEFAULT NULL,
   `update_time` datetime DEFAULT NULL,
+  `commit_time` datetime DEFAULT NULL,
+  `finish_time` datetime DEFAULT NULL,
+  `rollback_time` datetime DEFAULT NULL,
+  `owner` varchar(128) not null default '' comment '正在处理全局事务的锁定者',
   PRIMARY KEY (`id`),
   UNIQUE KEY `gid` (`gid`),
+  key `owner`(`owner`),
   KEY `create_time` (`create_time`),
   KEY `update_time` (`update_time`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
 
-drop table IF EXISTS saga_step;
-CREATE TABLE IF NOT EXISTS `saga_step` (
+drop table IF EXISTS trans_branch;
+CREATE TABLE IF NOT EXISTS `trans_branch` (
   `id` int(11) NOT NULL AUTO_INCREMENT,
-  `gid` varchar(45) NOT NULL COMMENT '事务全局id',
-  `data` json DEFAULT NULL COMMENT '请求所携带的数据',
-  `step` int(11) NOT NULL COMMENT '处于saga中的第几步',
+  `gid` varchar(128) NOT NULL COMMENT '事务全局id',
   `url` varchar(128) NOT NULL COMMENT '动作关联的url',
-  `type` varchar(45) NOT NULL COMMENT 'saga的所有步骤',
-  `status` varchar(45) NOT NULL COMMENT '步骤的状态 prepared | succeed | failed',
+  `data` TEXT COMMENT '请求所携带的数据',
+  `branch` VARCHAR(128) NOT NULL COMMENT '事务分支名称',
+  `branch_type` varchar(45) NOT NULL COMMENT '事务分支类型 saga_action | saga_compensate | xa',
+  `status` varchar(45) NOT NULL COMMENT '步骤的状态 committed | finished | rollbacked',
   `finish_time` datetime DEFAULT NULL,
   `rollback_time` datetime DEFAULT NULL,
   `create_time` datetime DEFAULT NULL,
   `update_time` datetime DEFAULT NULL,
   PRIMARY KEY (`id`),
-  UNIQUE KEY `gid` (`gid`,`step`),
+  UNIQUE KEY `gid` (`gid`,`branch`, `branch_type`),
   KEY `create_time` (`create_time`),
   KEY `update_time` (`update_time`)
-) ENGINE=InnoDB AUTO_INCREMENT=17 DEFAULT CHARSET=utf8mb4;
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
 
 drop table IF EXISTS trans_log;
 CREATE TABLE IF NOT EXISTS `trans_log` (
   `id` int(11) NOT NULL AUTO_INCREMENT,
-  `gid` varchar(45) NOT NULL COMMENT '事务全局id',
-  `step` int(11) DEFAULT NULL COMMENT 'saga的步骤',
+  `gid` varchar(128) NOT NULL COMMENT '事务全局id',
+  `branch` varchar(128) DEFAULT NULL COMMENT '事务分支',
   `action` varchar(45) DEFAULT NULL COMMENT '行为',
-  `status` varchar(45) NOT NULL COMMENT '状态',
-  `detail` json NOT NULL COMMENT '行为之后的status',
+  `old_status` varchar(45) NOT NULL DEFAULT '' COMMENT '旧状态',
+  `new_status` varchar(45) NOT NULL COMMENT '新状态',
+  `detail` TEXT NOT NULL COMMENT '行为记录的数据',
   `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
   PRIMARY KEY (`id`),
   KEY `gid` (`gid`),
   KEY `create_time` (`create_time`)
-) ENGINE=InnoDB AUTO_INCREMENT=48 DEFAULT CHARSET=utf8mb4;
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
 

dtmsvr/main.go

@@ -1,11 +1,15 @@
 package dtmsvr
 
 import (
+	"fmt"
+
 	"github.com/sirupsen/logrus"
 	"github.com/yedf/dtm/common"
 	"github.com/yedf/dtm/examples"
 )
 
+var dtmsvrPort = 8080
+
 func Main() {
 	go StartSvr()
 }
@@ -15,8 +19,8 @@ func StartSvr() {
 	common.InitApp(common.GetCurrentPath(), &config)
 	app := common.GetGinApp()
 	AddRoute(app)
-	logrus.Printf("dtmsvr listen at: 8080")
-	app.Run(":8080")
+	logrus.Printf("dtmsvr listen at: %d", dtmsvrPort)
+	app.Run(fmt.Sprintf(":%d", dtmsvrPort))
 }
 
 func PopulateMysql() {

examples/conf.yml.sample

@@ -0,0 +1,6 @@
+Mysql:
+  host: 'mysql'
+  user: 'root'
+  password: 'my-secret-pw'
+  database: 'dtm'
+  port: '3306'

examples/data.go

@@ -21,7 +21,7 @@ func RunSqlScript(mysql map[string]string, script string) {
 	sqls := strings.Split(string(content), ";")
 	for _, sql := range sqls {
 		s := strings.TrimSpace(sql)
-		if strings.Contains(strings.ToLower(s), "drop") || s == "" {
+		if strings.Contains(strings.ToLower(s), "drop1") || s == "" {
 			continue
 		}
 		logrus.Printf("executing: '%s'", s)

examples/examples.sql

@@ -1,7 +1,7 @@
-CREATE DATABASE `dtm_busi` /*!40100 DEFAULT CHARACTER SET utf8mb4 */;
+CREATE DATABASE if not exists `dtm_busi` /*!40100 DEFAULT CHARACTER SET utf8mb4 */;
 use dtm_busi;
 drop table if exists user_account;
-create table user_account(
+create table if not exists user_account(
   id int(11) PRIMARY KEY AUTO_INCREMENT,
   user_id int(11) UNIQUE ,
   balance DECIMAL(10, 2) not null default '0',
@@ -11,4 +11,4 @@ create table user_account(
   key(update_time)
 );
 
-insert into user_account (user_id, balance) values (1, 10000), (2, 10000);
+insert into user_account (user_id, balance) values (1, 10000), (2, 10000) on DUPLICATE KEY UPDATE balance=values (balance);