配置 Saga Server · BeetlSQL3 官方使用手冊

### 配置 Saga Server Saga Server負責管理回滾任務，Saga-server的實現是基于SpringBoot，并內置H2數據庫。在啟動Saga-Server前，需要安裝和啟動Kafka。進入命令行，直接啟動Saga-Server ``` java -jar sql-saga-microservice-server-3.1.5-RELEASE.jar ``` 默認情況下，會鏈接本地的`127.0.0.1:9092`的kafka，并且，使用H2數據庫，數據存放在~/.h2目錄下你也可以通過SpringBoot機制，配置kafka 和數據庫，如下配置項 ```properties server.port=18081 server.shutdown=graceful #數據庫配置 spring.datasource.url=jdbc:h2:file:~/.h2/saga-server;AUTO_SERVER=TRUE spring.datasource.username=sa spring.datasource.password= spring.datasource.driver-class-name=org.h2.Driver # kafka配置 spring.kafka.bootstrapServers=${kafka_server:127.0.0.1:9092} spring.kafka.consumer.group-id=${kafka_group:saga-group} spring.kafka.consumer.max-poll-records=100 spring.kafka.consumer.auto-offset-reset=latest spring.kafka.consumer.enable-auto-commit=true spring.kafka.consumer.auto-commit-interval=100 spring.kafka.listener.type=batch beetlsql-saga.kafka.server-topic=saga-server-topic beetlsql-saga.kafka.client-topic-prefix=saga-client beetlsql.sqlManagers=mySqlManager beetlsql.mySqlManager.ds=datasource beetlsql.mySqlManager.basePackage=org.beetl.sql.saga.ms.server beetlsql.mySqlManager.dbStyle=org.beetl.sql.core.db.H2Style ``` 關于Kakfa Topic說明 * beetlsql-saga.kafka.server-topic， server監聽的topic，客戶端產生的回滾任務或者回滾結果都發送到此topic * beetlsql-saga.kafka.client-topic-prefix ，客戶端topic前綴，比如客戶端是orderApp，那么，Saga-Server會發送orderApp相關的回滾任務到saga-client-orderApp里。每個客戶端都有一個topic接受來自Saga—Server的回滾任務。對于每個客戶端，安裝方式類似上一章的多庫實現 ```java <dependency> <groupId>com.ibeetl</groupId> <artifactId>sql-springboot-starter</artifactId> <version>${version}</version> </dependency> <dependency> <groupId>com.ibeetl</groupId> <artifactId>sql-saga-client</artifactId> <version>${version}</version> </dependency> ``` client包自帶了`SagaClientConfig` 用于設置`SagaContext.sagaContextFactory = new SagaClientContextFactory(this);` 因此基于Spring Boot的客戶端必須掃描包`org.beetl.sql.saga.ms.client`,比如 ```java @SpringBootApplication(scanBasePackages = {"com.xxx", "org.beetl.sql.saga.ms.client"}) @EnableKafka public class UserApplication { public static void main(String[] args) { SpringApplication.run(UserApplication.class, args); } } ``` > 可以參考源碼DemoApplication來了解如何配置和使用Saga Client 客戶端需要配置appName以及跟Saga-Server交互的Topic，如下 ```properties spring.application.name=demoSystem beetlsql-saga.kafka.client-topic-prefix=saga-client beetlsql-saga.kafka.server-topic=saga-server-topic spring.kafka.bootstrapServers=${kafka_server:127.0.0.1:9092} spring.kafka.consumer.max-poll-records=1 spring.kafka.consumer.auto-offset-reset=latest spring.kafka.consumer.enable-auto-commit=true spring.kafka.listener.type=batch ``` 第一行是標準的Spring Boot 配置，后面倆行跟Saga-Server一樣的配置完成如上配置后，既可以同樣的方式使用BeetlSQL的Saga事務管理，以源碼`DemoController`為例子 ```java SagaContext sagaContext = SagaContext.sagaContextFactory.current(); try { sagaContext.start(gid); //模擬調用倆個微服務，訂單和用戶 rest.postForEntity(orderAddUrl, null,String.class, paras); rest.postForEntity(userBalanceUpdateUrl, null,String.class, paras); if (1 == 1) { throw new RuntimeException("模擬失敗,查詢saga-server 看效果"); } } catch (Exception e) { sagaContext.rollback(); return e.getMessage(); } ``` orderAddUrl 是訂單服務，實現如下 ```java @Service public class OrderService { @Autowired OrderMapper orderMapper; @Transactional(propagation=Propagation.NEVER) public void addOrder(String orderId,String userId,Integer fee){ SagaContext sagaContext = SagaContext.sagaContextFactory.current(); try{ sagaContext.start(orderId); OrderEntity orderEntity = new OrderEntity(); orderEntity.setFee(fee); orderEntity.setUserId(userId); orderEntity.setProductId("any"); orderMapper.insert(orderEntity); sagaContext.commit(); }catch (Exception e){ sagaContext.rollback(); throw new RuntimeException(e); } } } ``` userBalanceUpdateUrl是余額操作，實現如下 ```java @Service public class UserService { @Autowired UserMapper userMapper; @Transactional(propagation= Propagation.NEVER) public void update(String orderId,String userId,Integer fee){ SagaContext sagaContext = SagaContext.sagaContextFactory.current(); try{ sagaContext.start(orderId); UserEntity user = userMapper.unique(userId); user.setBalance(user.getBalance()-fee); userMapper.updateById(user); sagaContext.commit(); }catch (Exception e){ sagaContext.rollback(); throw new RuntimeException(e); } } } ```