智牛股_第3章_交易平台_ gRPC+Seata Server分布式事务

智牛股_第3章_交易平台_ gRPC+Seata Server分布式事务

交易平台 - Day 3

学习目标

目标1： Restful HATEOAS的使用

目标2： gRPC基本概念与工作机制

目标3： gRPC使用

目标4： Seata Server分布式事务原理

目标5： Fescar使用

第1章 Restful介绍

1. 目标

• 了解Restful基本概念： 掌握定义， 基本功能， 使用场景。

2. 分析

• Restful定义
• Richardson成熟模型
• 常用HTTP状态码
• 良好的URI编写规范

3. 讲解

3.1 Restful定义

Rest是一种软件架构与设计风格， 并非一套标准， 只提供了一些原则与约定条件。

RESTful提供了一组架构约束，当作为一个整体服务来应用时，强调组件交互的可伸缩性、

接口的通用性、组件的独⽴部署、以及用来减少交互延迟、增强安全性、封装遗留系统的中间组件。

满足这些约束条件和原则的应用程序或设计就是Restful。

3.2 Richardson成熟模型

由Leonard Richardson发明的Rest成熟模型：

等级2加入了HTTP方法处理：

等级3为超媒体控制(HATEOAS)， 也是最为成熟的Rest模型。

3.3 常用HTTP状态码

3.4 良好的URI规范

• URI的路径采用斜杠分隔符（/）来表示资源之间的层次关系。
• URI的路径使用逗号（,）与分号(;)来表示非层次元素。
• URI的查询部分，使用与符号（&）来分割参数。
• URI中避免出现文件扩展名（例：.jsp、.json、.xml、.html）

4. 总结

• 理解Richardson成熟模型的进阶层次， 常见HTTP状态码的含义，形成良好的URI规范。

第2章 HATEOAS介绍

1. 目标

• 了解HATEOAS基本概念与用法
• HATEOAS在项目工程中的具体实现与应用

2. 步骤

• HATEOAS简述
• HATEOAS示例
• HATEOAS常用链接类型
• 服务设计
• 工程结构说明
• 启动股票服务验证
• 启动订单服务验证

3. 讲解

3.1 HATEOAS简述

HATEOAS（Hypermedia as the engine of application state）是 REST 架构风格中最复杂的约束，也是构建成熟 REST 服务的核心。

3.2 HATEOAS示例

3.3 HATEOAS常用链接类型

<list>
  <device>
    ......
    <link rel="self" href="http://host:port/res/device/11"/>
  device>
  ...
  <link rel="next" href="http://host:port/res/device?start=10&size=10"/>
list>
1
2
3
4
5
6
7
8

3.4 HATEOAS CRUD示例

• 显示接口

  http://39.98.152.160:10680/admin/accountWarnNotifyTemplate/search

• 分页查询

  http://39.98.152.160:10680/admin/accountWarnNotifyTemplate {?page,size,sort}

支持排序：

http://39.98.152.160:10680/admin/accountWarnNotifyTemplate?page=0&size=3&sort=lastUpdateUser,desc

• 新增数据

  http://39.98.152.160:10680/admin/accountWarnNotifyTemplate

  传递的操作类型要改为POST， 采用JSON格式提交数据。

• 更新数据

  http://39.98.152.160:10680/admin/accountWarnNotifyTemplate/15

  链接附带数据唯一ID， 提交采用PUT方式。

• 删除数据

  http://39.98.152.160:10680/admin/accountWarnNotifyTemplate/15

  提交方式采用PUT。

3.5 服务设计

采用Spring Data Rest 实现 Hypermedia规范。

设计两个服务， 订单服务和股票服务， 两个服务遵循Hateoas风格。

• Step 1: 通过Restful的Hypermedia模型调用股票服务， 查询并打印股票信息。
• Step 2: 通过HTTP PUT动作更新股票价格。
• Step 3: 重新调用股票信息接口，打印股票名称与价格。
• Step 4: 以上步骤操作成功后， 订单服务调用自身接口， 生成订单信息。

3.6 工程说明

数据层采用spring data jpa，spring提供的一套简化JPA开发的框架，按照约定好的【方法命名规则】写dao层接口，就可以在不写接口实现的情况下，实现对数据库的访问和操作。同时提供了很多除了CRUD之外的功能，如分页、排序、复杂查询等等。

本工程侧重Hateoas的理解， 数据库采用简化的H2内存数据库， 重新启动服务数据消失。

1. hateoas-demo父级工程
   父级工程， 负责子工程依赖与打包管理。
   POM依赖：

 <dependencies>
        
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-webartifactId>
        dependency>
		
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-data-jpaartifactId>
        dependency>
        
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-data-restartifactId>
        dependency>
        <dependency>
            <groupId>org.jadira.usertypegroupId>
            <artifactId>usertype.coreartifactId>
            <version>6.0.1.GAversion>
        dependency>
        
        <dependency>
            <groupId>com.fasterxml.jackson.datatypegroupId>
            <artifactId>jackson-datatype-hibernate5artifactId>
            <version>2.9.8version>
        dependency>
        
        <dependency>
            <groupId>com.fasterxml.jackson.dataformatgroupId>
            <artifactId>jackson-dataformat-xmlartifactId>
            <version>2.9.0version>
        dependency>
		
        <dependency>
            <groupId>org.apache.commonsgroupId>
            <artifactId>commons-lang3artifactId>
        dependency>
		
        <dependency>
            <groupId>com.h2databasegroupId>
            <artifactId>h2artifactId>
        dependency>
        
        <dependency>
            <groupId>org.projectlombokgroupId>
            <artifactId>lombokartifactId>
            <optional>trueoptional>
        dependency>
    dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

2. hateoas-stocks股票服务工程

• HateoasStocksApplication启动类：

@SpringBootApplication()
@ComponentScan(basePackages = {"com.itcast"})
@EntityScan(basePackages = {"com.itcast"})
@EnableJpaRepositories(basePackages =  {"com.itcast"})
@EnableCaching
public class HateoasStocksApplication {

    public static void main(String[] args) {

        SpringApplication.run(HateoasStocksApplication.class, args);
    }

    @Bean
    public Hibernate5Module hibernate5Module() {
        return new Hibernate5Module();
    }

    @Bean
    public Jackson2ObjectMapperBuilderCustomizer jacksonBuilderCustomizer() {
        return builder -> {
            builder.indentOutput(true);
        };
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

注意： 要加上EntityScan与EnableJpaRepositories注解，指定路径， 否则不生效。

• StockRepository代码：

  定义两个方法，根据名称集合查找多个股票信息； 根据指定名称查找股票信息。

  按照JPA规范，按照方法名称自动映射解析， 无须写SQL。

@RepositoryRestResource(path = "/stocks")
public interface StockRepository extends JpaRepository<StocksEntity, Long> {
    
  /**
   * 根据股票名称查找所对应的股票数据
   * @param list
   * @return
   */
  List<StocksEntity> findByNameInOrderById(@Param("list")List<String> list);
    
  /**
   * 根据名称查询股票信息
   * @param name
   * @return
   */  
  public StocksEntity findByName(@Param("name")String name);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3. hateoas-order订单服务工程

HateoasOrderApplication启动类：

 @SpringBootApplication
 @ComponentScan(basePackages = {"com.itcast"})
 @EntityScan(basePackages = {"com.itcast"})
 @EnableJpaRepositories(basePackages =  {"com.itcast"})
 public class HateoasOrderApplication {
 
     public static void main(String[] args) {
         SpringApplication.run(HateoasOrderApplication.class, args);
     }
 
     /**
      * 采用JACKSON作为JSON处理组件
      * @return
      */
     @Bean
     public Jackson2HalModule jackson2HalModule() {
         return new Jackson2HalModule();
     }
 
     /**
      * 设置HTTP连接池参数
      * @return
      */
     @Bean
     public HttpComponentsClientHttpRequestFactory requestFactory() {
         PoolingHttpClientConnectionManager connectionManager =
                 new PoolingHttpClientConnectionManager(30, TimeUnit.SECONDS);
         connectionManager.setMaxTotal(200);
         connectionManager.setDefaultMaxPerRoute(20);
 
         CloseableHttpClient httpClient = HttpClients.custom()
                 .setConnectionManager(connectionManager)
                 .evictIdleConnections(30, TimeUnit.SECONDS)
                 .disableAutomaticRetries()
                 //  Keep-Alive 策略
                 .setKeepAliveStrategy(new RemoteConnectionKeepAliveStrategy())
                 .build();
 
         HttpComponentsClientHttpRequestFactory requestFactory =
                 new HttpComponentsClientHttpRequestFactory(httpClient);
 
         return requestFactory;
     }
 
     /**
      * 设置RestTemplate参数
      * @param builder
      * @return
      */
     @Bean
     public RestTemplate restTemplate(RestTemplateBuilder builder) {
         return builder
                 .setConnectTimeout(Duration.ofMillis(2000))
                 .setReadTimeout(Duration.ofMillis(1800))
                 .requestFactory(this::requestFactory)
                 .build();
     }
 
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

RemoteConnectionKeepAliveStrategy类代码：

public class RemoteConnectionKeepAliveStrategy implements org.apache.http.conn.ConnectionKeepAliveStrategy {
    private final long DEFAULT_SECONDS = 30;

    /**
     ** 远程连接， keepalive的设置策略
     */ 
    @Override
    public long getKeepAliveDuration(HttpResponse response, HttpContext context) {
        return Arrays.asList(response.getHeaders(HTTP.CONN_KEEP_ALIVE))
                .stream()
                .filter(h -> StringUtils.endsWithIgnoreCase(h.getName(), "timeout")
                        && StringUtils.isNumeric(h.getValue()))
                .findFirst()
                .map(h -> NumberUtils.toLong(h.getValue(), DEFAULT_SECONDS))
                .orElse(DEFAULT_SECONDS) * 1000;
    }
}
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

核心处理类RemoteRunner实现ApplicationRunner接口， 系统启动成功后便会执行run方法:

@Component
@Slf4j
public class RemoteRunner implements ApplicationRunner {
    private static final URI ROOT_URI = URI.create("http://localhost:8080/");

    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private OrderRepository orderRepository;

    @Override
    public void run(ApplicationArguments args) throws Exception {

        Link stocksLink = getLink(ROOT_URI,"stocksEntities");
        // Step 1: 查询股票信息
        queryStocks(stocksLink);
        // Step 2: 更新股票价格
        Link updateLink= getLink(ROOT_URI.resolve("stocks/1"),"stocksEntity");
        Resource<StocksEntity> americano = updateStocks(updateLink);
        // Step 3: 重新查询打印股票信息
        queryStocks(stocksLink);

        // Step 4: 生成订单信息
        OrderEntity order = OrderEntity.builder()
                .user("mirson")
                .stockName("建设银行")
                .volume(1000)
                .price(99.9)
                .build();
        orderRepository.save(order);

    }

    /**
     * 获取请求链接
     * @param uri
     * @param rel
     * @return
     */
    private Link getLink(URI uri, String rel) {
        ResponseEntity<Resources<Link>> rootResp =
                restTemplate.exchange(uri, HttpMethod.GET, null,
                        new ParameterizedTypeReference<Resources<Link>>() {});
        Link link = rootResp.getBody().getLink(rel);
        log.info("Link: {}", link);
        return link;
    }

    /**
     * 查询股票信息
     * @param stocksLink
     */
    private void queryStocks(Link stocksLink) {
        ResponseEntity<PagedResources<Resource<StocksEntity>>> stocksResp =
                restTemplate.exchange(stocksLink.getTemplate().expand(),
                        HttpMethod.GET, null,
                        new ParameterizedTypeReference<PagedResources<Resource<StocksEntity>>>() {});
        if(null != stocksResp.getBody() && null != stocksResp.getBody().getContent() ) {
            StringBuffer strs = new StringBuffer();
            stocksResp.getBody().getContent().forEach((s)->{
                strs.append(s.getContent().getName()).append(":").append(s.getContent().getPrice()).append( ",");
            });
            String resp = strs.toString().replaceAll(",$", "");
            log.info("query stocks ==> " + resp);
        }else {
            log.info("query stocks ==>  empty! ");
        }

    }

    /**
     * 更新股票信息
     * @param link
     * @return
     */
    private Resource<StocksEntity> updateStocks(Link link) {

        StocksEntity americano = StocksEntity.builder()
                .name("中国平安")
                .price(68.9)
                .build();
        RequestEntity<StocksEntity> req =
                RequestEntity.put(link.getTemplate().expand()).body(americano);
        ResponseEntity<Resource<StocksEntity>> resp =
                restTemplate.exchange(req,
                        new ParameterizedTypeReference<Resource<StocksEntity>>() {});
        log.info("add Stocks ==> {}", resp);
        return resp.getBody();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

3.7 启动股票服务验证

1. 启动股票服务，通过HTTP访问， 来查看Rest接口信息

   • 地址： http://127.0.0.1:8080/

     {
       "_links" : {
         "stocksEntities" : {
           "href" : "http://127.0.0.1:8080/stocks{?page,size,sort}",
           "templated" : true
         },
         "profile" : {
           "href" : "http://127.0.0.1:8080/profile"
         }
       }
     }
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11

     可以看到我们定义的/stocks接口

   • 地址： http://127.0.0.1:8080/stocks

     {
       "_embedded" : {
         "stocksEntities" : [ {
           "createTime" : "2019-07-09T14:20:44.644+0000",
           "updateTime" : "2019-07-09T14:20:44.644+0000",
           "name" : "中国平安",
           "price" : 68.6,
           "_links" : {
             "self" : {
               "href" : "http://127.0.0.1:8080/stocks/1"
             },
             "stocksEntity" : {
               "href" : "http://127.0.0.1:8080/stocks/1"
             }
           }
         }, {
           "createTime" : "2019-07-09T14:20:44.647+0000",
           "updateTime" : "2019-07-09T14:20:44.647+0000",
           "name" : "工商银行",
           "price" : 58.8,
           "_links" : {
             "self" : {
               "href" : "http://127.0.0.1:8080/stocks/2"
             },
             "stocksEntity" : {
               "href" : "http://127.0.0.1:8080/stocks/2"
             }
           }
         }, {
           "createTime" : "2019-07-09T14:20:44.648+0000",
           "updateTime" : "2019-07-09T14:20:44.648+0000",
           "name" : "招商银行",
           "price" : 98.9,
           "_links" : {
             "self" : {
               "href" : "http://127.0.0.1:8080/stocks/3"
             },
             "stocksEntity" : {
               "href" : "http://127.0.0.1:8080/stocks/3"
             }
           }
         } ]
       },
       "_links" : {
         "self" : {
           "href" : "http://127.0.0.1:8080/stocks{?page,size,sort}",
           "templated" : true
         },
         "profile" : {
           "href" : "http://127.0.0.1:8080/profile/stocks"
         },
         "search" : {
           "href" : "http://127.0.0.1:8080/stocks/search"
         }
       },
       "page" : {
         "size" : 20,
         "totalElements" : 3,
         "totalPages" : 1,
         "number" : 0
       }
     }
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30
     31
     32
     33
     34
     35
     36
     37
     38
     39
     40
     41
     42
     43
     44
     45
     46
     47
     48
     49
     50
     51
     52
     53
     54
     55
     56
     57
     58
     59
     60
     61
     62

     打印了所有股票信息，最下面还暴露了其他接口信息。

   • 地址： http://127.0.0.1:8080/stocks/search

     {
       "_links" : {
         "findByName" : {
           "href" : "http://127.0.0.1:8080/stocks/search/findByName{?name}",
           "templated" : true
         },
         "findByNameInOrderById" : {
           "href" : "http://127.0.0.1:8080/stocks/search/findByNameInOrderById{?list}",
           "templated" : true
         },
         "self" : {
           "href" : "http://127.0.0.1:8080/stocks/search"
         }
       }
     }
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15

     打印了我们通过JPA定义的两个接口，findByNameInOrderById与findByName。

   • 地址： http://127.0.0.1:8080/stocks/search/findByNameInOrderById?list=中国平安,test

     查询中国平安与test两只股票，结果：

     {
       "_embedded" : {
         "stocksEntities" : [ {
           "createTime" : "2019-07-09T14:20:44.644+0000",
           "updateTime" : "2019-07-09T14:20:44.644+0000",
           "name" : "中国平安",
           "price" : 68.6,
           "_links" : {
             "self" : {
               "href" : "http://127.0.0.1:8080/stocks/1"
             },
             "stocksEntity" : {
               "href" : "http://127.0.0.1:8080/stocks/1"
             }
           }
         } ]
       },
       "_links" : {
         "self" : {
           "href" : "http://127.0.0.1:8080/stocks/search/findByNameInOrderById?list=%E4%B8%AD%E5%9B%BD%E5%B9%B3%E5%AE%89,latte"
         }
       }
     }
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23

3.8 启动订单服务验证

1. 启动订单服务， 完整验证四个步骤处理流程。

   条件： 订单服务， 预期是修改中国平安的股票， 从价格68.6改成68.9；

   新增订单信息： 股票名称建设银行，用户名是mirson, 交易数量为1000， 价格为99.9。

2. 控制台日志

看到两行日志，中国平安的价格发生了变化，从68.6修改为68.9。

3. 查看order服务的订单信息

   地址： http://127.0.0.1:8082/order

   {
     "_embedded" : {
       "orderEntities" : [ {
         "createTime" : "2019-07-09T14:35:42.520+0000",
         "updateTime" : "2019-07-09T14:35:42.520+0000",
         "user" : "mirson",
         "stockName" : "建设银行",
         "volume" : 1000,
         "price" : 99.9,
         "_links" : {
           "self" : {
             "href" : "http://127.0.0.1:8082/order/1"
           },
           "orderEntity" : {
             "href" : "http://127.0.0.1:8082/order/1"
           }
         }
       } ]
     },
     "_links" : {
       "self" : {
         "href" : "http://127.0.0.1:8082/order{?page,size,sort}",
         "templated" : true
       },
       "profile" : {
         "href" : "http://127.0.0.1:8082/profile/order"
       },
       "search" : {
         "href" : "http://127.0.0.1:8082/order/search"
       }
     },
     "page" : {
       "size" : 20,
       "totalElements" : 1,
       "totalPages" : 1,
       "number" : 0
     }
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38

   生成了我们新增的订单信息。此外还可以通过Postman来模拟增删改查操作， Spring Data Rest都帮我们做好封装。

   通过Spring Data Rest整个实现流程非常简单， 没有Controller层， 这正是Restful的设计风格， 以资源为对象， 无需过多的流程，转换处理。

4. 总结

• 理解整体服务结构设计， 如何整合Spring Data Rest 实现 Hypermedia规范。
• 如何编写HATEOAS服务， 客户端的调用实现方式。

第3章 GRPC介绍与使用

1. 目标

• 了解GRPC基本概念，工作机制与特性
• GRPC在项目中整合运用， 实现一个基本的GRPC服务调用。

2. 分析

• GRPC简介
• GRPC特性
• GRPC线程模型
• 客户端调用流程
• 服务设计
• 工程结构，搭建配置
• Protoc生成Java文件
• 服务接口实现
• 启动验证

3. 讲解

3.1 GPRC简介

gRPC 是Google开源的高性能、通用的RPC框架。客户端与服务端约定接口调用， 可以在各种环境中运行，具有跨语言特性， 适合构建分布式、微服务应用。

3.2 GPRC特性

• 性能优异：

  1. 采用Proto Buffer作序列化传输， 对比JSON与XML有数倍提升。

  2. 采用HTTP2协议， 头部信息压缩， 对连接进行复用， 减少TCP连接次数。

  3. gRPC底层采用Netty作为NIO处理框架， 提升性能。

• 多语言支持，多客户端接入， 支持C++/GO/Ruby等语言。

• 支持负载均衡、跟踪、健康检查和认证。

3.3 GPRC线程模型

gRPC 的线程模型遵循 Netty 的线程分工原则，协议层消息的接收和编解码由 Netty 的 I/O(NioEventLoop) 线程负责, 应用层的处理由应用线程负责，防止由于应用处理耗时而阻塞 Netty 的 I/O 线程。

BIO线程模型采用了线程池，但是后端的应用处理线程仍然采用同步阻塞的模型，阻塞的时间取决对方I/O处理的速度和网络I/O传输的速度。

采用线程池模式的BIO:

NIO 线程模型(Reactor模式):

3.4 客户端调用流程

1. 客户端 Stub 调用 发起 RPC 调用 远程服务。
2. 获取服务端的地址信息（列表），使用默认的 LoadBalancer 策略，选择一个具体的 gRPC 服务端。
3. 如果与服务端之间没有可用的连接，则创建 NettyClientTransport 和 NettyClientHandler，建立 HTTP/2 连接。
4. 对请求使用 PB（Protobuf）序列化，通过 HTTP/2 Stream 发送给 gRPC 服务端。
5. 服务端接收到响应之后，使用 PB（Protobuf）做反序列化。
6. 回调 GrpcFuture 的 set(Response) 方法，唤醒阻塞的客户端调用线程，获取 RPC 响应数据。

3.5 GRpc vs Rest 性能对比

1. GRpc与Rest性能对比

   在不同操作系统平台, 不同请求数的对比:
   

2. GRpc + ProtoBuf 与Rest(Http+Json)性能对比

   Go项目测试地址

实测结果显示GRpc的通讯方案, 性能有32%的提升, 资源占用降低30%左右。

3.6 服务设计

3.7 工程结构

|- grpc-demo
|-- grpc-client
|-- grpc-lib
|-- grpc-server

• grpc-demo : 父级工程， 管理依赖相关。
• grpc-client: 客户端工程，负责调用gRPC服务, 提供HTTP服务触发。
• grpc-server: 股票服务端工程， 提供股票价格接口。
• grpc-lib: 公用工程，生成为protobuf对象与gRPC Service。

3.8 Protoc编译工具

下载工具： protoc

下载插件： protoc-gen-grpc-java

3.9 工程说明

1. grpc-demo

   POM文件

     <dependencies>
          
           <dependency>
               <groupId>net.devhgroupId>
               <artifactId>grpc-server-spring-boot-starterartifactId>
               <version>2.5.0.RELEASEversion>
           dependency>
           
           <dependency>
               <groupId>net.devhgroupId>
               <artifactId>grpc-client-spring-boot-starterartifactId>
               <version>2.5.0.RELEASEversion>
           dependency>
       dependencies>
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14

2. grpc-lib公用组件工程

   StockService.proto文件：

   syntax = "proto3";
   
   option java_multiple_files = true;
   option java_package = "com.itcast.grpc.lib";
   option java_outer_classname = "StockServiceProto";
   
   // The stock service definition.
   service StockService {
       // get stock price by name
       rpc GetStockPrice (StockServiceRequest) returns (StockServiceReply) {
       }
   }
   
   // The request message
   message StockServiceRequest {
       string name = 1;
   }
   
   // The response message
   message StockServiceReply {
       string message = 1;
   }
   
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23

   POM依赖

   <dependencies>
            
           <dependency>
               <groupId>com.google.protobufgroupId>
               <artifactId>protobuf-javaartifactId>
               <version>3.8.0version>
           dependency>
           <dependency>
               <groupId>io.grpcgroupId>
               <artifactId>grpc-netty-shadedartifactId>
               <version>1.22.1version>
           dependency>
           
           <dependency>
               <groupId>io.grpcgroupId>
               <artifactId>grpc-protobufartifactId>
               <version>1.22.1version>
               <exclusions>
                   <exclusion>
                       <artifactId>protobuf-javaartifactId>
                       <groupId>com.google.protobufgroupId>
                   exclusion>
               exclusions>
           dependency>
           <dependency>
               <groupId>io.grpcgroupId>
               <artifactId>grpc-stubartifactId>
               <version>1.22.1version>
           dependency>
   dependencies>
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30

   注意Protobuf生成工具和组件要保持一致， 工具我们用的是3.8.0最新版， 依赖也要改成3.8.0，排除了grpc-protobuf的传递依赖。

进入proto文件目录， 执行以下命令， 生成gRPC对象与Service:

d:/TestCode/protoc.exe --plugin=protoc-gen-grpc-java=d:/TestCode/protoc-grpc.exe --java_out=./ --grpc-java_out=./ StockService.proto
1

注意插件路径要写正确， 要指定protobuf与grpc两个输出位置， 命令指定在当前同级目录生成协议文件。

3. grpc-server服务端
   GrpcStockService类， 重写gRPC Service定义的接口，生成指定范围随机数的价格 ：

   @GrpcService
   public class GrpcStockService extends StockServiceGrpc.StockServiceImplBase {
   
       @Override
       public void getStockPrice(StockServiceRequest request, StreamObserver<StockServiceReply> responseObserver) {
           String msg = "股票名称:" + request.getName() + ", 股票价格:" + (new Random().nextInt(100-20)+20);
           StockServiceReply reply =StockServiceReply.newBuilder().setMessage(msg).build();
           responseObserver.onNext(reply);
           responseObserver.onCompleted();
       }
   }
   
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12

   GrpcServiceStartup启动类：

   @SpringBootApplication
   @ComponentScan(basePackages = {"com.itcast"})
   public class GrpcServerStartup {
    public static void main(String[] args) {
        SpringApplication.run(GrpcServerStartup.class, args);
    }
   }
   
   
   1
   2
   3
   4
   5
   6
   7
   8
   9

4. grpc-client客户端

   GrpcClientService类：

   @Service
   public class GrpcClientService {
   
       @GrpcClient("grpc-server")
       private StockServiceGrpc.StockServiceBlockingStub stockServiceStub;
   
       public String getStockPrice(final String name) {
           try {
               final StockServiceReply response = stockServiceStub.getStockPrice(StockServiceRequest.newBuilder().setName(name).build());
               return response.getMessage();
           } catch (final StatusRuntimeException e) {
               return "error!";
           }
       }
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15

   注解GrpcClient映射的名称为grpc-server， 不能随便填写，要与配置保持一致。
   GrpcClientApplication启动类：

   @SpringBootApplication
   @RestController
   @ComponentScan(basePackages = {"com.itcast"})
   public class GrpcClientApplication {
   
       @Autowired
       private GrpcClientService grpcClientService;
   
       public static void main(String[] args) {
           SpringApplication.run(GrpcClientApplication.class, args);
       }
   
       @RequestMapping("/")
       public String getStockPrice(@RequestParam(defaultValue = "中国平安") String name) {
           return grpcClientService.getStockPrice(name);
       }
   }  
   
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18

   application.yml配置文件：

   server:
     port: 9000
   spring:
     application:
       name: grpc-client
   
   grpc:
     client:
       grpc-server:
         address: 'static://127.0.0.1:9999'
         enableKeepAlive: true
         keepAliveWithoutCalls: true
         negotiationType: plaintext
   
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14

   这里面定义名为【grpc-server】的服务配置信息， 与上面注解要保持一致。

3.10 启动验证

启动服务端与客户端， 访问客户端地址： http://127.0.0.1:9000/getStockPrice?name=中国银行

不断刷新请求， 股票价格也会随机变化，能够正常结合Spring Boot访问gRPC服务。

4. 总结

• 着重了解GRPC的特性， 性能优势， 客户端的调用处理流程。
• 掌握GRPC服务的集成与配置， 实现服务端的配置与客户端的调用。

第4章 SEATA 介绍与使用

1. 目标

• 了解分布式事务解决方案， 基本理论
• 了解SEATA SERVER的处理机制， 功能作用
• 掌握SEATA 与工程项目的整合
• 掌握SEATA 的配置与使用

2. 分析

• SEATA简介
• 传统分布式事务解决方案
• CAP理论与BASE理论
• SEATA工作处理机制
• 服务设计
• 工程结构与搭建配置
• 服务启动， 功能验证

3. 讲解

3.1 SEATA简介

Seata前身叫Fescar， 2019 年 1 月，阿里巴巴中间件团队发起了开源项目 Fescar（Fast & EaSy Commit And Rollback），和社区一起共建开源分布式事务解决方案。Fescar 的愿景是让分布式事务的使用像本地事务的使用一样，简单和高效，并且能够逐步解决开发者们遇到的分布式事务方面的所有难题。

Fescar 开源后，蚂蚁金服加入 Fescar 社区参与共建，并在 Fescar 0.4.0 版本中贡献了 TCC 模式。为了打造更中立、更开放、生态更加丰富的分布式事务开源社区，经过社区核心成员的投票，大家决定对 Fescar 进行品牌升级，并更名为 Seata，意为：Simple Extensible Autonomous Transaction Architecture，中文直译就是：简单的、可扩展的、自治的事务架构。是一套一站式分布式事务解决方案。

Seata文档地址

发展历程：

3.2 传统分布式事务解决方案

• 两阶段提交（2PC)
  

  两阶段事务提交长时间锁定， 但也不能保证事务的百分百可靠，同时对性能较大影响，某个服务出现故障， 影响全局事务， 可用性差，不适合分布式微服务领域。

• 补偿事务（TCC)

​ 主要分为Try、Confirm、Cancel三个阶段。

​ Try 主要做检测校验及预处理工作；

​ Confirm 是对业务做确认提交动作， 一般Try处理成功， Confirm也会成功。

​ Cancel是在某个业务环节执行错误的时候， 或者极端Confirm出错情况下， 执行的补偿方法。比如转账没有成功到达对方账户， 那么Cancel就要把钱退回转帐方账户。

​ TCC侵入性较强， 需要写较多补偿方法， 加入补偿机制， 而且必须保障幂等，整体复杂， 且开发量大， 也不易维护。

• 异步消息一致性

​ 将分布式事务拆分成本地事务， 通过消息队列记录并通知各服务事务处理结果：

​

1. A 系统先发送一个 prepared 消息到 mq，如果这个 prepared 消息发送失败那么就直接取消操作别执行了；

2. 如果这个消息发送成功了，那么接着执行本地事务，如果成功就告诉 mq 发送确认消息，如果失败就告诉 mq 回滚消息；

3. 如果发送了确认消息，那么此时 B 系统会接收到确认消息，然后执行本地的事务；

4. 消息队列会自动定时轮询所有发送过 prepared 消息但未发送确认消息的服务，这个消息是不是本地事务处理失败了， 是继续重试还是回滚？服务可以查下数据库看之前本地事务是否执行，如果回滚了，那么这里也回滚吧。这个机制的作用就是避免可能本地事务执行成功了，而确认消息却发送失败了。

5. 这个方案里，要是系统 B 的事务失败了咋办？自动不断重试直到成功，如果实在是不行，要么就是针对重要的资金类业务进行全局回滚，比如 B 系统本地回滚后，再通知系统 A 也回滚；或是发送报警由人工来回滚或补偿。

3.3 CAP理论

CAP的含义：

• C：Consistency 一致性
• A：Availability 可用性
• P：Partition tolerance 分区容错性
  在分布式系统中，C、A、P三个条件中我们最多只能满足两个要求。
  一般在分布式领域， 会通过牺牲一致性来换取系统的可用性和分区容错性。

3.4 BASE理论

所谓的“牺牲一致性”并不是完全放弃数据一致性，而是牺牲强一致性换取弱一致性，这样我们就有兼顾全局的可能。BASE理论：

• BA：Basic Available 基本可用
  整个系统在某些不可抗力的情况下，仍然能够保证“可用性”，即一定时间内仍然能够返回一个明确的结果。只不过“基本可用”和“高可用”的区别是：
  【一定时间】可以适当延长 ；
  当举行大促时，【响应时间】可以适当延长；
  给部分用户直接返回一个降级页面，从而缓解服务器压力。但要注意，返回降级页面仍然是返回明确结果。
• S：Soft State：柔性状态
  同一数据的不同副本的状态，可以不需要实时一致。
• E：Eventual Consisstency：最终一致性
  同一数据的不同副本的状态，可以不需要实时一致，但一定要保证经过一定时间后仍然是一致的。

3.5 SEATA处理机制

所有微服务以本地事务方式处理，作为分支事务， 各服务之间通过RPC通信， 所有分支事务由全局事务管控。

SEATA分布式事务的解决方案是由一个全局事务（Global Transaction）， 和一批分支事务（Branch Transation）组成， 分支事务也就是各微服务的本地事务。

SEATA的三大组件：

• Transaction Coordinator(TC)：维护全局和分支事务的状态，驱动全局事务提交与回滚。

• Transaction Manager™：定义全局事务的范围：开始、提交或回滚全局事务。

• Resource Manager(RM)：管理分支事务处理的资源，与 TC 通信 并注册分支事务, 然后报告分支事务的状态，最后驱动分支事务提交或回滚。

SEATA分布式事务管理的生命周期过程：

1. TM 要求 TC 开始新的全局事务，TC 生成表示全局事务的 XID。

2. XID 通过微服务的调用链传播。

3. RM 在 TC 中将本地事务注册为 XID 的相应全局事务的分支。

4. TM 要求 TC 提交或回滚 XID 的相应全局事务。

5. TC 驱动 XID 的相应全局事务下的所有分支事务，完成分支提交或回滚。

更多机制实现细节及原理，比如AT模式、MT模式和XA模式的原理与设计， 请参考SEATA WIKI

3.6 服务设计

这里引用官方的与Spring Boot的结合示例，Seata-samples-springboot 集成到我们工程当中， 并做调整改进。

主要设计了两个接口， 一个是修改状态， 另外一个是修改数量，并抛出异常， 两个接口没有受内部事务控制， 集成在一个工程当中。

增加数量接口会模拟抛出异常， 我们需要验证， 修改状态接口的数据是否会产生回滚。

3.7 工程结构

官方DEMO实现有些仓促，存在些bug， 依赖较为混乱， 集成了DUBBO， 重新做了调整， 去除DUBBO， 简化依赖， 集成到我们工程当中便于演示。

• SeataDemoApplication启动类：

@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@ComponentScan(basePackages = {"com.itcast"})
@EntityScan(basePackages = {"com.itcast"})
@EnableJpaRepositories(basePackages =  {"com.itcast"})
@RestController
@EnableTransactionManagement
public class SeataDemoApplication {

    final String ASSET_ID = "14070e0e3cfe403098fa9ca37e8e7e76";

    @Autowired
    private IAssignService assignService;

    public static void main(String[] args) {

        SpringApplication.run(SeataDemoApplication.class, args);
    }

    /**
     * Home string.
     *
     * @return the string
     */
    @RequestMapping(value = "/asset/assign")
    @ResponseBody
    public String assetAssign() {

        String result;
        try {
            AssetAssign assetAssign = assignService.increaseAmount(
                    ASSET_ID);
            result = assetAssign.toString();
        } catch (Exception e) {
            result = ExceptionUtils.getMessage(e);

        }
        return result;
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

注意要开启EnableJpaRepositories和ComponentScan包扫描。

这里定义了外部接口， 调用内部service方法。

• AssetServiceImpl类：

@Service
@Component
public class AssetServiceImpl implements IAssetService {

    /**
     * The constant LOGGER.
     */
    public static final Logger LOGGER = LoggerFactory.getLogger(IAssetService.class);

    /**
     * The constant ASSET_ID.
     */
    public static final String ASSET_ID = "DF001";

    @Autowired
    private AssetRepository assetRepository;

    @Override
    public int increase() {
        LOGGER.info("Asset Service Begin ... xid: " + RootContext.getXID() + "\n");
        Asset asset = assetRepository.findById(ASSET_ID).get();
        asset.setAmount(asset.getAmount().add(new BigDecimal("1")));
        assetRepository.save(asset);
        throw new RuntimeException("test exception for seata, your transaction should be rollbacked,asset=" + asset);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

这里定义了一个增加数量的接口， 内部会抛出一个自定义的RuntimeException, 这个方法没有加任何事务处理。

• AssignServiceImpl类：

@Service
public class AssignServiceImpl implements IAssignService {
	private static final Logger LOGGER = LoggerFactory.getLogger(AssignServiceImpl.class);

	@Autowired
	private AssignRepository assignRepository;

	@Autowired
    private IAssetService assetService;

	@Override
	@Transactional
	@GlobalTransactional
	public AssetAssign increaseAmount(String id) {
		LOGGER.info("Assign Service Begin ... xid: " + RootContext.getXID() + "\n");
		AssetAssign assetAssign = assignRepository.findById(id).get();
		assetAssign.setStatus("2");
		assignRepository.save(assetAssign);

		// remote call asset service
		assetService.increase();
		return assetAssign;
	}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

这里实现两个动作， 一个是修改assetAssign的状态， 另一个是调用增加数量的接口。
注意官方示例里面@Transactional是开启的(但没有配置rollback回滚)， 因为JPA需要在事务内进行操作。
GlobalTransactional是Seata提供的注解， 这个需要加上。
从这两个业务实现可以看到， 内部没有采用事务处理，如果没有纳入分布式事务， 即便抛出异常， 对数据库的操作仍会生效。

• SeataConfiguration配置类：

/**
 * The type Fescar configuration.
 */
@Configuration
public class SeataConfiguration {

	@Value("${spring.application.name}")
	private String applicationId;

	/**
	 * 注册一个扫描器, 扫描全局分布式事务
	 *
	 * @return global transaction scanner
	 */
	@Bean
	public GlobalTransactionScanner globalTransactionScanner() {
        GlobalTransactionScanner globalTransactionScanner = new GlobalTransactionScanner(applicationId,
            "my_test_tx_group");
		return globalTransactionScanner;
	}
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

里面指定了一个名称为my_test_tx_group组别， 这是一个标识， Seata可以支持多个组别， 多个配置存在。

• 工程配置
  application.yml：

  server:
      port: 9999
      servlet:
          context-path: /demo
  spring:
      application:
          name: seata-springboot-app
      datasource:
          driverClassName: com.mysql.jdbc.Driver
          url: jdbc:mysql://192.168.19.150:3306/seata?useSSL=false&serverTimezone=UTC
          username: root
          password: 654321
          poolPingConnectionsNotUsedFor: 60000
          removeAbandoned: true
          connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
          minIdle: 1
          validationQuery: SELECT 1 FROM DUAL
          initialSize: 5
          maxWait: 60000
          poolPreparedStatements: false
          filters: stat,wall
          testOnBorrow: false
          testWhileIdle: true
          minEvictableIdleTimeMillis: 300000
          timeBetweenEvictionRunsMillis: 60000
          testOnReturn: false
          maxActive: 50
          druid:
              user: admin
              password: admin
  
      jpa: 
          hibernate:
              ddl-auto: none
          show-sql: true
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35

  file.conf(Seata配置)

  transport {
    # tcp udt unix-domain-socket
    type = "TCP"
    #NIO NATIVE
    server = "NIO"
    #enable heartbeat
    heartbeat = true
    #thread factory for netty
    thread-factory {
      boss-thread-prefix = "NettyBoss"
      worker-thread-prefix = "NettyServerNIOWorker"
      server-executor-thread-prefix = "NettyServerBizHandler"
      share-boss-worker = false
      client-selector-thread-prefix = "NettyClientSelector"
      client-selector-thread-size = 1
      client-worker-thread-prefix = "NettyClientWorkerThread"
      # netty boss thread size,will not be used for UDT
      boss-thread-size = 1
      #auto default pin or 8
      worker-thread-size = 8
    }
  }
  store {
    # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
    max-branch-session-size = 16384
    # globe session size , if exceeded throws exceptions
    max-global-session-size = 512
    # file buffer size , if exceeded allocate new buffer
    file-write-buffer-cache-size = 16384
    # when recover batch read size
    session.reload.read_size = 100
  }
  service {
    #vgroup->rgroup
    vgroup_mapping.my_test_tx_group = "default"
    #only support single node
    default.grouplist = "127.0.0.1:8091"
    #degrade current not support
    enableDegrade = false
    #disable
    disable = false
  }
  
  client {
    async.commit.buffer.limit = 10000
    lock {
      retry.internal = 10
      retry.times = 30
    }
  }
  
  ## transaction log store
  store {
    ## store mode: file、db
    mode = "file"
  
    ## file store
    file {
      dir = "file_store/data"
  
      # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
      max-branch-session-size = 16384
      # globe session size , if exceeded throws exceptions
      max-global-session-size = 512
      # file buffer size , if exceeded allocate new buffer
      file-write-buffer-cache-size = 16384
      # when recover batch read size
      session.reload.read_size = 100
    }
  
    ## database store
    db {
      driver_class = ""
      url = ""
      user = ""
      password = ""
    }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52
  53
  54
  55
  56
  57
  58
  59
  60
  61
  62
  63
  64
  65
  66
  67
  68
  69
  70
  71
  72
  73
  74
  75
  76
  77
  78

  registry.conf（Seata配置):

  registry {
    # file 、nacos 、eureka、redis、zk
    type = "file"
  
    file {
      name = "file.conf"
    }
  }
  
  config {
    # file、nacos 、apollo、zk
    type = "file"
  
    file {
      name = "file.conf"
    }
  }
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18

3.8 启动验证

启动服务之前，先要启动Seata-Server。 可直接下载运行包

如果采用ZK配置， 下载运行Zookeeper

确保数据库init_db.sql脚本执行成功， 如果seata采用数据库模式， 确保db_store.sql文件执行成功。

1. 启动服务，

2. 访问地址： http://127.0.0.1:9999/demo/asset/assign

抛出一个异常， 控制台日志， 打印出回滚信息：

3. 验证数据库
   查看Asset表
   

   amout数量还是为1， 回滚成功。
   查看Asset_assign表
   

   status为04， 没有被修改成2， 回滚成功。

4. 总结

• 着重掌握分布式事务解决方案， 知道SEATA的工作处理机制， 如何解决分布式事务问题。
• 从服务设计中， 通过多个服务的不同接口操作， 来验证分布式事务的一致性，掌握Seata的整合配置，支持多种方式配置， 根据实际情况结合应用。
• Seata侵入性低， 接入使用比较简单， 在实际应用中， 如果出现问题，也要多查看Seata Server的服务运行日志。