devbox
weixin_40725706
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

spring cloud alibaba开发笔记十三（订单微服务，openFeign，seate的实践应用）_seate pom依赖

作者：weixin_40725706 | 2024-02-29 12:50:08

赞

踩

seate pom依赖

首先创建订单微服务e-commerce-order-service

pom

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>e-commerce-service</artifactId>
  7.         <groupId>com.taluohui.ecommerce</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  11.  
  12.     <artifactId>e-commerce-order-service</artifactId>
  13.     <version>1.0-SNAPSHOT</version>
  14.     <packaging>jar</packaging>
  15.  
  16.     <!-- 模块名及描述信息 -->
  17.     <name>e-commerce-order-service</name>
  18.     <description>订单服务</description>
  19.  
  20.     <properties>
  21.         <maven.compiler.source>8</maven.compiler.source>
  22.         <maven.compiler.target>8</maven.compiler.target>
  23.     </properties>
  24.  
  25.     <dependencies>
  26.         <!-- spring cloud alibaba nacos discovery 依赖 -->
  27.         <dependency>
  28.             <groupId>com.alibaba.cloud</groupId>
  29.             <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
  30.         </dependency>
  31.         <!-- zipkin = spring-cloud-starter-sleuth + spring-cloud-sleuth-zipkin-->
  32.         <dependency>
  33.             <groupId>org.springframework.cloud</groupId>
  34.             <artifactId>spring-cloud-starter-zipkin</artifactId>
  35.         </dependency>
  36.         <dependency>
  37.             <groupId>org.springframework.kafka</groupId>
  38.             <artifactId>spring-kafka</artifactId>
  39.             <version>2.5.0.RELEASE</version>
  40.         </dependency>
  41.         <!-- Java Persistence API, ORM 规范 -->
  42.         <dependency>
  43.             <groupId>org.springframework.boot</groupId>
  44.             <artifactId>spring-boot-starter-data-jpa</artifactId>
  45.         </dependency>
  46.         <!-- MySQL 驱动, 注意, 这个需要与 MySQL 版本对应 -->
  47.         <dependency>
  48.             <groupId>mysql</groupId>
  49.             <artifactId>mysql-connector-java</artifactId>
  50.             <version>8.0.12</version>
  51.             <scope>runtime</scope>
  52.         </dependency>
  53.         <!-- SpringCloud Stream + Kafka -->
  54.         <dependency>
  55.             <groupId>org.springframework.cloud</groupId>
  56.             <artifactId>spring-cloud-stream</artifactId>
  57.         </dependency>
  58.         <dependency>
  59.             <groupId>org.springframework.cloud</groupId>
  60.             <artifactId>spring-cloud-stream-binder-kafka</artifactId>
  61.         </dependency>
  62.         <dependency>
  63.             <groupId>com.taluohui.ecommerce</groupId>
  64.             <artifactId>e-commerce-service-config</artifactId>
  65.             <version>1.0-SNAPSHOT</version>
  66.         </dependency>
  67.         <dependency>
  68.             <groupId>com.taluohui.ecommerce</groupId>
  69.             <artifactId>e-commerce-service-sdk</artifactId>
  70.             <version>1.0-SNAPSHOT</version>
  71.         </dependency>
  72.         <!-- 集成 hystrix -->
  73.         <dependency>
  74.             <groupId>org.springframework.cloud</groupId>
  75.             <artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
  76.         </dependency>
  77.     </dependencies>
  78.  
  79.     <!--
  80.         SpringBoot的Maven插件, 能够以Maven的方式为应用提供SpringBoot的支持，可以将
  81.         SpringBoot应用打包为可执行的jar或war文件, 然后以通常的方式运行SpringBoot应用
  82.      -->
  83.     <build>
  84.         <finalName>${artifactId}</finalName>
  85.         <plugins>
  86.             <plugin>
  87.                 <groupId>org.springframework.boot</groupId>
  88.                 <artifactId>spring-boot-maven-plugin</artifactId>
  89.                 <executions>
  90.                     <execution>
  91.                         <goals>
  92.                             <goal>repackage</goal>
  93.                         </goals>
  94.                     </execution>
  95.                 </executions>
  96.             </plugin>
  97.         </plugins>
  98.     </build>
  99. </project>

配置文件

  1. server:
  2.   port: 8002
  3.   servlet:
  4.     context-path: /ecommerce-order-service
  5.  
  6. spring:
  7.   main:
  8.     allow-bean-definition-overriding: true #允许同名的Bean
  9.   application:
  10.     name: e-commerce-order-service # 应用名称也是构成 Nacos 配置管理 dataId 字段的一部分 (当 config.prefix 为空时)
  11.   cloud:
  12.     stream:
  13.       kafka:
  14.         binder:
  15.           brokers: 1.15.247.9:9092
  16.           auto-create-topics: true
  17.       bindings:
  18.         logisticsOutput:
  19.           destination: e-commerce-topic   # kafka topic
  20.           content-type: text/plain
  21.     alibaba:
  22.       seata:
  23.         tx-service-group: imooc-ecommerce # seata 全局事务分组
  24.     nacos:
  25.       # 服务注册发现
  26.       discovery:
  27.         enabled: true # 如果不想使用 Nacos 进行服务注册和发现, 设置为 false 即可
  28.         server-addr: 1.15.247.9:8848
  29.         # server-addr: 127.0.0.1:8848,127.0.0.1:8849,127.0.0.1:8850 # Nacos 服务器地址
  30.         namespace: 22d40198-8462-499d-a7fe-dbb2da958648
  31.         metadata:
  32.           management:
  33.             context-path: ${server.servlet.context-path}/actuator
  34.     # 开启 ribbon 重试机制, 即获取服务失败是否从另外一个节点重试
  35.     loadbalancer:
  36.       retry:
  37.         enabled: true
  38.   kafka:
  39.     bootstrap-servers: 1.15.247.9:9092
  40.     producer:
  41.       retries: 3
  42.     consumer:
  43.       auto-offset-reset: latest
  44.   sleuth:
  45.     sampler:
  46.       # ProbabilityBasedSampler 抽样策略
  47.       probability: 1.0  # 采样比例, 1.0 表示 100%, 默认是 0.1
  48.       # RateLimitingSampler 抽样策略, 设置了限速采集, spring.sleuth.sampler.probability 属性值无效
  49.       rate: 100  # 每秒间隔接受的 trace 量
  50.   zipkin:
  51.     sender:
  52.       type: kafka # 默认是 web
  53.     base-url: http://1.15.247.9:9411/
  54.   jpa:
  55.     show-sql: true
  56.     hibernate:
  57.       ddl-auto: none
  58.     properties:
  59.       hibernate.show_sql: true
  60.       hibernate.format_sql: true
  61.     open-in-view: false
  62.   datasource:
  63.     # 数据源
  64.     url: jdbc:mysql://1.15.247.9:3306/ecommerce?autoReconnect=true&useUnicode=true&characterEncoding=utf8&useSSL=false
  65.     username: root
  66.     password: Cjw970404
  67.     type: com.zaxxer.hikari.HikariDataSource
  68.     driver-class-name: com.mysql.cj.jdbc.Driver
  69.     # 连接池
  70.     hikari:
  71.       maximum-pool-size: 8
  72.       minimum-idle: 4
  73.       idle-timeout: 30000
  74.       connection-timeout: 30000
  75.       max-lifetime: 45000
  76.       auto-commit: true
  77.       pool-name: ImoocEcommerceHikariCP
  78.  
  79. # feign 相关的配置
  80. feign:
  81.   hystrix:
  82.     enabled: true   # Hystrix 默认是关闭的
  83.   client:
  84.     config:
  85.       default:  # 全局的
  86.         connectTimeout: 2000  # 默认的连接超时时间是 10s
  87.         readTimeout: 5000
  88.  
  89. # 暴露端点
  90. management:
  91.   endpoints:
  92.     web:
  93.       exposure:
  94.         include: '*'
  95.   endpoint:
  96.     health:
  97.       show-details: always

入口启动类

  1. /**
  2.  * <h1>订单微服务启动入口</h1>
  3.  * */
  4. @EnableJpaAuditing
  5. @SpringBootApplication
  6. @EnableCircuitBreaker
  7. @EnableFeignClients
  8. @EnableDiscoveryClient
  9. public class OrderApplication {
  10.  
  11.     public static void main(String[] args) {
  12.  
  13.         SpringApplication.run(OrderApplication.class, args);
  14.     }
  15. }

订单的sql，这边只是简单的演示，还有很多字段，可以自行添加

  1. -- 创建 t_ecommerce_order 数据表
  2. CREATE TABLE IF NOT EXISTS `ecommerce`.`t_ecommerce_order` (
  3.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增主键',
  4.   `user_id` bigint(20) NOT NULL DEFAULT 0 COMMENT '用户 id',
  5.   `address_id` bigint(20) NOT NULL DEFAULT 0 COMMENT '用户地址记录 id',
  6.   `order_detail` text NOT NULL COMMENT '订单详情(json 存储, goodsId, count)',
  7.   `create_time` datetime NOT NULL DEFAULT '0000-01-01 00:00:00' COMMENT '创建时间',
  8.   `update_time` datetime NOT NULL DEFAULT '0000-01-01 00:00:00' COMMENT '更新时间',
  9.   PRIMARY KEY (`id`)
  10. ) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=utf8 COMMENT='用户订单表';

entity

  1. /**
  2.  * <h1>用户订单表实体类定义</h1>
  3.  * */
  4. @Data
  5. @NoArgsConstructor
  6. @AllArgsConstructor
  7. @Entity
  8. @EntityListeners(AuditingEntityListener.class)
  9. @Table(name = "t_ecommerce_order")
  10. public class EcommerceOrder {
  11.  
  12.     /** 自增主键 */
  13.     @Id
  14.     @GeneratedValue(strategy = GenerationType.IDENTITY)
  15.     @Column(name = "id", nullable = false)
  16.     private Long id;
  17.  
  18.     /** 用户 id */
  19.     @Column(name = "user_id", nullable = false)
  20.     private Long userId;
  21.  
  22.     /** 用户地址 id */
  23.     @Column(name = "address_id", nullable = false)
  24.     private Long addressId;
  25.  
  26.     /** 订单详情(json 存储) */
  27.     @Column(name = "order_detail", nullable = false)
  28.     private String orderDetail;
  29.  
  30.     /** 创建时间 */
  31.     @CreatedDate
  32.     @Column(name = "create_time", nullable = false)
  33.     private Date createTime;
  34.  
  35.     /** 更新时间 */
  36.     @LastModifiedDate
  37.     @Column(name = "update_time", nullable = false)
  38.     private Date updateTime;
  39.  
  40.     public EcommerceOrder(Long userId, Long addressId, String orderDetail) {
  41.  
  42.         this.userId = userId;
  43.         this.addressId = addressId;
  44.         this.orderDetail = orderDetail;
  45.     }
  46. }

vo

  1. /**
  2.  * <h1>订单详情</h1>
  3.  * */
  4. @ApiModel(description = "分页订单详情对象")
  5. @Data
  6. @NoArgsConstructor
  7. @AllArgsConstructor
  8. public class PageSimpleOrderDetail {
  9.     @ApiModelProperty(value = "订单详情")
  10.     private List<SingleOrderItem> orderItems;
  11.  
  12.     @ApiModelProperty(value = "是否有更多的订单(分页)")
  13.     private Boolean hasMore;
  14.  
  15.     /**
  16.      * <h2>单个订单信息</h2>
  17.      * */
  18.     @ApiModel(description = "单个订单信息对象")
  19.     @Data
  20.     @NoArgsConstructor
  21.     @AllArgsConstructor
  22.     public static class SingleOrderItem {
  23.  
  24.         @ApiModelProperty(value = "订单表主键 id")
  25.         private Long id;
  26.  
  27.         @ApiModelProperty(value = "用户地址信息")
  28.         private UserAddress userAddress;
  29.  
  30.         @ApiModelProperty(value = "订单商品信息")
  31.         private List<SingleOrderGoodsItem> goodsItems;
  32.     }
  33.  
  34.     @ApiModel(description = "单个订单中的单项商品信息")
  35.     @Data
  36.     @NoArgsConstructor
  37.     @AllArgsConstructor
  38.     public static class SingleOrderGoodsItem {
  39.  
  40.         @ApiModelProperty(value = "简单商品信息")
  41.         private SimpleGoodsInfo simpleGoodsInfo;
  42.  
  43.         @ApiModelProperty(value = "商品个数")
  44.         private Integer count;
  45.     }
  46. }

以及用于微服务之间传递的类

  1. /**
  2.  * <h1>订单信息</h1>
  3.  */
  4. @ApiModel(description = "用户发起购买订单")
  5. @Data
  6. @NoArgsConstructor
  7. @AllArgsConstructor
  8. public class OrderInfo {
  9.  
  10.     @ApiModelProperty(value = "用户地址表主键 id")
  11.     private Long userAddress;
  12.  
  13.     @ApiModelProperty(value = "订单中的商品信息")
  14.     private List<OrderItem> orderItems;
  15.  
  16.     /**
  17.      * <h2>订单中的商品信息</h2>
  18.      * */
  19.     @ApiModel(description = "订单中的单项商品信息")
  20.     @Data
  21.     @NoArgsConstructor
  22.     @AllArgsConstructor
  23.     public static class OrderItem {
  24.  
  25.         @ApiModelProperty(value = "商品表主键 id")
  26.         private Long goodsId;
  27.  
  28.         @ApiModelProperty(value = "购买商品个数")
  29.         private Integer count;
  30.  
  31.         public DeductGoodsInventory toDeductGoodsInventory() {
  32.             return new DeductGoodsInventory(this.goodsId, this.count);
  33.         }
  34.     }
  35. }

dao

  1. /**
  2.  * <h1>EcommerceOrder Dao 接口定义</h1>
  3.  * */
  4. public interface EcommerceOrderDao extends PagingAndSortingRepository<EcommerceOrder, Long> {
  5.  
  6.     /**
  7.      * <h2>根据 userId 查询分页订单</h2>
  8.      * select * from t_ecommerce_order where user_id = ?
  9.      *  order by ... desc/asc limit x offset y
  10.      * */
  11.     Page<EcommerceOrder> findAllByUserId(Long userId, Pageable pageable);
  12. }

service接口

  1. /**
  2.  * <h1>订单相关服务接口定义</h1>
  3.  * */
  4. public interface IOrderService {
  5.  
  6.     /**
  7.      * <h2>下单(分布式事务): 创建订单 -> 扣减库存 -> 扣减余额 -> 创建物流信息(Stream + Kafka)</h2>
  8.      * */
  9.     TableId createOrder(OrderInfo orderInfo);
  10.  
  11.     /**
  12.      * <h2>获取当前用户的订单信息: 带有分页</h2>
  13.      * */
  14.     PageSimpleOrderDetail getSimpleOrderDetailByPage(int page);
  15. }

openFeign的使用

由于订单微服务需要依赖其他微服务，所以需要使用到openFeign，进行微服务之间的通信。这里的openFeign只是简化使用，需要更多的设置，可以看前面专门写的openfeign。

由于所有的微服务都会做权限的校验，所以需要对openFeign进行配置，让他的请求头带上校验信息。配置文件如下。（在order微服务下创建feign文件夹，用于openFeign的使用）

  1. /**
  2.  * <h1>Feign 调用时, 把 Header 也传递到服务提供方</h1>
  3.  * */
  4. @Slf4j
  5. @Configuration
  6. public class FeignConfig {
  7.  
  8.     /**
  9.      * <h2>给 Feign 配置请求拦截器</h2>
  10.      * RequestInterceptor 是我们提供给 open-feign 的请求拦截器, 把 Header 信息传递
  11.      * */
  12.     @Bean
  13.     public RequestInterceptor headerInterceptor() {
  14.  
  15.         return template -> {
  16.  
  17.             ServletRequestAttributes attributes =
  18.                     (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
  19.             if (null != attributes) {
  20.                 HttpServletRequest request = attributes.getRequest();
  21.                 Enumeration<String> headerNames = request.getHeaderNames();
  22.                 if (null != headerNames) {
  23.                     while (headerNames.hasMoreElements()) {
  24.                         String name = headerNames.nextElement();
  25.                         String values = request.getHeader(name);
  26.                         // 不能把当前请求的 content-length 传递到下游的服务提供方, 这明显是不对的
  27.                         // 请求可能一直返回不了, 或者是请求响应数据被截断
  28.                         if (!name.equalsIgnoreCase("content-length")) {
  29.                             // 这里的 template 就是 RestTemplate
  30.                             template.header(name, values);
  31.                         }
  32.                     }
  33.                 }
  34.             }
  35.         };
  36.     }
  37. }

不需要兜底策略的服务间通信

下面创建不需要兜底策略的服务间通信，并不是所有服务间通信都需要兜底策略，有一些没有完成的需要即使的报错，比如扣减用户余额，如果使用兜底策略，明显是不对的。

  1. /**
  2.  * <h1>用户账户服务 Feign 接口</h1>
  3.  * */
  4. @FeignClient(
  5.         contextId = "NotSecuredBalanceClient",
  6.         value = "e-commerce-account-service"
  7. )
  8. public interface NotSecuredBalanceClient {
  9.  
  10.     @RequestMapping(
  11.             value = "/ecommerce-account-service/balance/deduct-balance",
  12.             method = RequestMethod.PUT
  13.     )
  14.     CommonResponse<BalanceInfo> deductBalance(@RequestBody BalanceInfo balanceInfo);
  15. }

商品

  1. public interface NotSecuredGoodsClient {
  2.     /**
  3.      * <h2>根据 ids 扣减商品库存</h2>
  4.      * */
  5.     @RequestMapping(
  6.             value = "/ecommerce-goods-service/goods/deduct-goods-inventory",
  7.             method = RequestMethod.PUT
  8.     )
  9.     CommonResponse<Boolean> deductGoodsInventory(
  10.             @RequestBody List<DeductGoodsInventory> deductGoodsInventories);
  11.  
  12.     /**
  13.      * <h2>根据 ids 查询简单的商品信息</h2>
  14.      * */
  15.     @RequestMapping(
  16.             value = "/ecommerce-goods-service/goods/simple-goods-info",
  17.             method = RequestMethod.POST
  18.     )
  19.     CommonResponse<List<SimpleGoodsInfo>> getSimpleGoodsInfoByTableId(
  20.             @RequestBody TableId tableId);
  21. }

需要兜底策略的服务间通信

在feign文件夹下创建hystrix文件夹，用于存放兜底策略

地址微服务

  1. /**
  2.  * <h1>用户账户服务 Feign 接口(安全的)</h1>
  3.  * */
  4. @FeignClient(
  5.         contextId = "AddressClient",
  6.         value = "e-commerce-account-service",
  7.         fallback = AddressClientHystrix.class
  8. )
  9. public interface AddressClient {
  10.  
  11.     /**
  12.      * <h2>根据 id 查询地址信息</h2>
  13.      * */
  14.     @RequestMapping(
  15.             value = "/ecommerce-account-service/address/address-info-by-table-id",
  16.             method = RequestMethod.POST
  17.     )
  18.     CommonResponse<AddressInfo> getAddressInfoByTablesId(@RequestBody TableId tableId);
  19. }

兜底策略

  1. /**
  2.  * <h1>账户服务熔断降级兜底策略</h1>
  3.  * */
  4. @Slf4j
  5. @Component
  6. public class AddressClientHystrix implements AddressClient {
  7.     @Override
  8.     public CommonResponse<AddressInfo> getAddressInfoByTablesId(TableId tableId) {
  9.         log.error("[account client feign request error in order service] get address info" +
  10.                 "error: [{}]", JSON.toJSONString(tableId));
  11.         return new CommonResponse<>(
  12.                 -1,
  13.                 "[account client feign request error in order service]",
  14.                 new AddressInfo(-1L, Collections.emptyList())
  15.         );
  16.     }
  17. }

商品服务

  1. /**
  2.  * <h1>安全的商品服务 Feign 接口</h1>
  3.  * */
  4. @FeignClient(
  5.         contextId = "SecuredGoodsClient",
  6.         value = "e-commerce-goods-service",
  7.         fallback = GoodsClientHystrix.class
  8. )
  9. public interface SecuredGoodsClient {
  10.  
  11.     /**
  12.      * <h2>根据 ids 查询简单的商品信息</h2>
  13.      * */
  14.     @RequestMapping(
  15.             value = "/ecommerce-goods-service/goods/simple-goods-info",
  16.             method = RequestMethod.POST
  17.     )
  18.     CommonResponse<List<SimpleGoodsInfo>> getSimpleGoodsInfoByTableId(
  19.             @RequestBody TableId tableId);
  20. }

兜底策略

  1. /**
  2.  * <h1>商品服务熔断降级兜底</h1>
  3.  * */
  4. @Slf4j
  5. @Component
  6. public class GoodsClientHystrix implements SecuredGoodsClient {
  7.     @Override
  8.     public CommonResponse<List<SimpleGoodsInfo>> getSimpleGoodsInfoByTableId(TableId tableId) {
  9.         log.error("[goods client feign request error in order service] get simple goods" +
  10.                 "error: [{}]", JSON.toJSONString(tableId));
  11.         return new CommonResponse<>(
  12.                 -1,
  13.                 "[goods client feign request error in order service]",
  14.                 Collections.emptyList()
  15.         );
  16.     }
  17. }

Client(微服务)使用Seata(AT模式)的步骤

pom.xml 中引入依赖: spring-cloud-starter alibaba-seata, HikariCP

  1. <!--        seata-->
  2.         <dependency>
  3.             <groupId>com.alibaba.cloud</groupId>
  4.             <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
  5.         </dependency>
  6. <!--        注册Seata数据源需要的连接池-->
  7.         <dependency>
  8.             <groupId>com.zaxxer</groupId>
  9.             <artifactId>HikariCP</artifactId>
  10.             <optional>true</optional>
  11.         </dependency>

创建undo_ log 表(如果业务使用了多个数据库,每一个数据库都要有这张表)

  1. CREATE TABLE IF NOT EXISTS `ecommerce`.`undo_log` (
  2.   `id` bigint(20) NOT NULL AUTO_INCREMENT,
  3.   `branch_id` bigint(20) NOT NULL,
  4.   `xid` varchar(100) NOT NULL,
  5.   `context` varchar(128) NOT NULL,
  6.   `rollback_info` longblob NOT NULL,
  7.   `log_status` int(11) NOT NULL,
  8.   `log_created` datetime NOT NULL,
  9.   `log_modified` datetime NOT NULL,
  10.   `ext` varchar(100) DEFAULT NULL,
  11.   PRIMARY KEY (`id`),
  12.   UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
  13. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

配置事务分组（配置文件file.conf，registry.conf）

file.conf

  1. ## transaction log store, only used in seata-server
  2. store {
  3.   ## store mode: file、db、redis
  4.   mode = "db"
  5.  
  6.   ## file store property
  7.   file {
  8.     ## store location dir
  9.     dir = "sessionStore"
  10.     # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
  11.     maxBranchSessionSize = 16384
  12.     # globe session size , if exceeded throws exceptions
  13.     maxGlobalSessionSize = 512
  14.     # file buffer size , if exceeded allocate new buffer
  15.     fileWriteBufferCacheSize = 16384
  16.     # when recover batch read size
  17.     sessionReloadReadSize = 100
  18.     # async, sync
  19.     flushDiskMode = async
  20.   }
  21.  
  22.   ## database store property
  23.   db {
  24.     ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp)/HikariDataSource(hikari) etc.
  25.     datasource = "druid"
  26.     ## mysql/oracle/postgresql/h2/oceanbase etc.
  27.     dbType = "mysql"
  28.     driverClassName = "com.mysql.jdbc.Driver"
  29.     url = "jdbc:mysql://127.0.0.1:3306/seata?autoReconnect=true&useUnicode=true&characterEncoding=utf8&useSSL=false"
  30.     user = "root"
  31.     password = "root"
  32.     minConn = 5
  33.     maxConn = 100
  34.     globalTable = "global_table"
  35.     branchTable = "branch_table"
  36.     lockTable = "lock_table"
  37.     queryLimit = 100
  38.     maxWait = 5000
  39.   }
  40.  
  41.   ## redis store property
  42.   redis {
  43.     host = "127.0.0.1"
  44.     port = "6379"
  45.     password = ""
  46.     database = "0"
  47.     minConn = 1
  48.     maxConn = 10
  49.     maxTotal = 100
  50.     queryLimit = 100
  51.   }
  52.  
  53. }
  54. ##事务分组
  55. service {
  56.   vgroupMapping.imooc-ecommerce = "default"
  57.   default.grouplist = "127.0.0.1:8091"
  58. }
  59. client {
  60.   async.commit.buffer.limit = 10000
  61.   lock {
  62.     retry.internal = 10
  63.     retry.times = 30
  64.   }
  65. }

 注意:需要在MySQL中创建库和表

  1. CREATE DATABASE `seata`;
  2.  
  3. CREATE TABLE IF NOT EXISTS `seata`.`global_table`
  4. (
  5.     `xid`                       VARCHAR(128) NOT NULL,
  6.     `transaction_id`            BIGINT,
  7.     `status`                    TINYINT      NOT NULL,
  8.     `application_id`            VARCHAR(32),
  9.     `transaction_service_group` VARCHAR(32),
  10.     `transaction_name`          VARCHAR(128),
  11.     `timeout`                   INT,
  12.     `begin_time`                BIGINT,
  13.     `application_data`          VARCHAR(2000),
  14.     `gmt_create`                DATETIME,
  15.     `gmt_modified`              DATETIME,
  16.     PRIMARY KEY (`xid`),
  17.     KEY `idx_gmt_modified_status` (`gmt_modified`, `status`),
  18.     KEY `idx_transaction_id` (`transaction_id`)
  19. ) ENGINE = InnoDB
  20.   DEFAULT CHARSET = utf8;
  21.  
  22. -- the table to store BranchSession data
  23. CREATE TABLE IF NOT EXISTS `seata`.`branch_table`
  24. (
  25.     `branch_id`         BIGINT       NOT NULL,
  26.     `xid`               VARCHAR(128) NOT NULL,
  27.     `transaction_id`    BIGINT,
  28.     `resource_group_id` VARCHAR(32),
  29.     `resource_id`       VARCHAR(256),
  30.     `branch_type`       VARCHAR(8),
  31.     `status`            TINYINT,
  32.     `client_id`         VARCHAR(64),
  33.     `application_data`  VARCHAR(2000),
  34.     `gmt_create`        DATETIME,
  35.     `gmt_modified`      DATETIME,
  36.     PRIMARY KEY (`branch_id`),
  37.     KEY `idx_xid` (`xid`)
  38. ) ENGINE = InnoDB
  39.   DEFAULT CHARSET = utf8;
  40.  
  41. -- the table to store lock data
  42. CREATE TABLE IF NOT EXISTS `seata`.`lock_table`
  43. (
  44.     `row_key`        VARCHAR(128) NOT NULL,
  45.     `xid`            VARCHAR(96),
  46.     `transaction_id` BIGINT,
  47.     `branch_id`      BIGINT       NOT NULL,
  48.     `resource_id`    VARCHAR(256),
  49.     `table_name`     VARCHAR(32),
  50.     `pk`             VARCHAR(36),
  51.     `gmt_create`     DATETIME,
  52.     `gmt_modified`   DATETIME,
  53.     PRIMARY KEY (`row_key`),
  54.     KEY `idx_branch_id` (`branch_id`)
  55. ) ENGINE = InnoDB
  56.   DEFAULT CHARSET = utf8;

registry.conf

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

bootstrap.yml

  1. spring:
  2.   cloud:
  3.     alibaba:
  4.       seata:
  5.         tx-service-group: imooc-ecommerce # seata 全局事务分组

配置Seata数据源代理(思考下,为什么需要这个? )

  1. import com.zaxxer.hikari.HikariDataSource;
  2. import io.seata.rm.datasource.DataSourceProxy;
  3. import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  6. import org.springframework.context.annotation.Primary;
  7.  
  8. import javax.sql.DataSource;
  9.  
  10. /**
  11.  * <h1>Seata 所需要的数据源代理配置类</h1>
  12.  * */
  13. @Configuration
  14. public class DataSourceProxyAutoConfiguration {
  15.  
  16.     private final DataSourceProperties dataSourceProperties;
  17.  
  18.     public DataSourceProxyAutoConfiguration(DataSourceProperties dataSourceProperties) {
  19.         this.dataSourceProperties = dataSourceProperties;
  20.     }
  21.  
  22.     /**
  23.      * <h2>配置数据源代理, 用于 Seata 全局事务回滚</h2>
  24.      * before image + after image -> undo_log
  25.      * */
  26.     @Primary
  27.     @Bean("dataSource")
  28.     public DataSource dataSource() {
  29.  
  30.         HikariDataSource dataSource = new HikariDataSource();
  31.         dataSource.setJdbcUrl(dataSourceProperties.getUrl());
  32.         dataSource.setUsername(dataSourceProperties.getUsername());
  33.         dataSource.setPassword(dataSourceProperties.getPassword());
  34.         dataSource.setDriverClassName(dataSourceProperties.getDriverClassName());
  35.  
  36.         return new DataSourceProxy(dataSource);
  37.     }
  38. }

加载拦截器SeataHandlerInterceptor,实现微服务之间xid的传递

  1.     /**
  2.      * <h2>添加拦截器配置</h2>
  3.      * */
  4.     @Override
  5.     protected void addInterceptors(InterceptorRegistry registry) {
  6.  
  7.         // 添加用户身份统一登录拦截的拦截器
  8.         registry.addInterceptor(new LoginUserInfoInterceptor())
  9.                 .addPathPatterns("/**").order(0);
  10.         // Seata 传递 xid 事务 id 给其他的微服务
  11.         // 只有这样, 其他的服务才会写 undo_log, 才能够实现回滚
  12.         registry.addInterceptor(new SeataHandlerInterceptor()).addPathPatterns("/**");
  13.     }

将@GlobalTransactional注解标注在需要分布式事务的方法上

使用SpringCloud Stream完成消息通信

首先引入依赖，添加配置文件（在创建微服务时已经引入，看上面）

这里使用的是自定义通道，所以要创建自定义通道接口

  1. /**
  2.  * <h1>自定义物流消息通信信道(Source)</h1>
  3.  * */
  4. public interface LogisticsSource {
  5.  
  6.     /** 输出信道名称 */
  7.     String OUTPUT = "logisticsOutput";
  8.  
  9.     /**
  10.      * <h2>物流 Source -> logisticsOutput</h2>
  11.      * 通信信道的名称是 logisticsOutput, 对应到 yml 文件里的配置
  12.      * */
  13.     @Output(LogisticsSource.OUTPUT)
  14.     MessageChannel logisticsOutput();
  15. }

在公共服务内创建微服务之间传递的对象

  1. /**
  2.  * <h1>创建订单时发送的物流消息</h1>
  3.  * */
  4. @ApiModel(description = "Stream 物流消息对象")
  5. @Data
  6. @NoArgsConstructor
  7. @AllArgsConstructor
  8. public class LogisticsMessage {
  9.  
  10.     @ApiModelProperty(value = "用户表主键 id")
  11.     private Long userId;
  12.  
  13.     @ApiModelProperty(value = "订单表主键 id")
  14.     private Long orderId;
  15.  
  16.     @ApiModelProperty(value = "用户地址表主键 id")
  17.     private Long addressId;
  18.  
  19.     @ApiModelProperty(value = "备注信息(json 存储)")
  20.     private String extraInfo;
  21. }

结合上面所有的知识点，完成商品的下单和查询操作

  1. import com.alibaba.fastjson.JSON;
  2. import com.imooc.ecommerce.account.AddressInfo;
  3. import com.imooc.ecommerce.account.BalanceInfo;
  4. import com.imooc.ecommerce.common.TableId;
  5. import com.imooc.ecommerce.dao.EcommerceOrderDao;
  6. import com.imooc.ecommerce.entity.EcommerceOrder;
  7. import com.imooc.ecommerce.feign.AddressClient;
  8. import com.imooc.ecommerce.feign.NotSecuredBalanceClient;
  9. import com.imooc.ecommerce.feign.NotSecuredGoodsClient;
  10. import com.imooc.ecommerce.feign.SecuredGoodsClient;
  11. import com.imooc.ecommerce.filter.AccessContext;
  12. import com.imooc.ecommerce.goods.DeductGoodsInventory;
  13. import com.imooc.ecommerce.goods.SimpleGoodsInfo;
  14. import com.imooc.ecommerce.order.LogisticsMessage;
  15. import com.imooc.ecommerce.order.OrderInfo;
  16. import com.imooc.ecommerce.service.IOrderService;
  17. import com.imooc.ecommerce.source.LogisticsSource;
  18. import com.imooc.ecommerce.vo.PageSimpleOrderDetail;
  19. import io.seata.spring.annotation.GlobalTransactional;
  20. import lombok.extern.slf4j.Slf4j;
  21. import org.apache.commons.collections4.CollectionUtils;
  22. import org.springframework.beans.factory.annotation.Autowired;
  23. import org.springframework.cloud.stream.annotation.EnableBinding;
  24. import org.springframework.data.domain.Page;
  25. import org.springframework.data.domain.PageRequest;
  26. import org.springframework.data.domain.Pageable;
  27. import org.springframework.data.domain.Sort;
  28. import org.springframework.messaging.support.MessageBuilder;
  29. import org.springframework.stereotype.Service;
  30.  
  31. import java.util.ArrayList;
  32. import java.util.Collections;
  33. import java.util.HashSet;
  34. import java.util.List;
  35. import java.util.Map;
  36. import java.util.Set;
  37. import java.util.function.Function;
  38. import java.util.stream.Collectors;
  39.  
  40. /**
  41.  * <h1>订单相关服务接口实现</h1>
  42.  * */
  43. @Slf4j
  44. @Service
  45. @EnableBinding(LogisticsSource.class)
  46. public class OrderServiceImpl implements IOrderService {
  47.  
  48.     /** 表的 dao 接口 */
  49.     private final EcommerceOrderDao orderDao;
  50.  
  51.     /** Feign 客户端 */
  52.     private final AddressClient addressClient;
  53.     private final SecuredGoodsClient securedGoodsClient;
  54.     private final NotSecuredGoodsClient notSecuredGoodsClient;
  55.     private final NotSecuredBalanceClient notSecuredBalanceClient;
  56.  
  57.     /** SpringCloud Stream 的发射器 */
  58.     private final LogisticsSource logisticsSource;
  59.  
  60.     public OrderServiceImpl(EcommerceOrderDao orderDao,
  61.                             AddressClient addressClient,
  62.                             SecuredGoodsClient securedGoodsClient,
  63.                             NotSecuredGoodsClient notSecuredGoodsClient,
  64.                             NotSecuredBalanceClient notSecuredBalanceClient,
  65.                             LogisticsSource logisticsSource) {
  66.         this.orderDao = orderDao;
  67.         this.addressClient = addressClient;
  68.         this.securedGoodsClient = securedGoodsClient;
  69.         this.notSecuredGoodsClient = notSecuredGoodsClient;
  70.         this.notSecuredBalanceClient = notSecuredBalanceClient;
  71.         this.logisticsSource = logisticsSource;
  72.     }
  73.  
  74.     /**
  75.      * <h2>创建订单: 这里会涉及到分布式事务</h2>
  76.      * 创建订单会涉及到多个步骤和校验, 当不满足情况时直接抛出异常;
  77.      * 1. 校验请求对象是否合法
  78.      * 2. 创建订单
  79.      * 3. 扣减商品库存
  80.      * 4. 扣减用户余额
  81.      * 5. 发送订单物流消息 SpringCloud Stream + Kafka
  82.      * */
  83.     @Override
  84.     @GlobalTransactional(rollbackFor = Exception.class)
  85.     public TableId createOrder(OrderInfo orderInfo) {
  86.  
  87.         // 获取地址信息
  88.         AddressInfo addressInfo = addressClient.getAddressInfoByTablesId(
  89.                 new TableId(Collections.singletonList(
  90.                         new TableId.Id(orderInfo.getUserAddress())))).getData();
  91.  
  92.         // 1. 校验请求对象是否合法(商品信息不需要校验, 扣减库存会做校验)
  93.         if (CollectionUtils.isEmpty(addressInfo.getAddressItems())) {
  94.             throw new RuntimeException("user address is not exist: "
  95.                     + orderInfo.getUserAddress());
  96.         }
  97.  
  98.         // 2. 创建订单
  99.         EcommerceOrder newOrder = orderDao.save(
  100.                 new EcommerceOrder(
  101.                         AccessContext.getLoginUserInfo().getId(),
  102.                         orderInfo.getUserAddress(),
  103.                         JSON.toJSONString(orderInfo.getOrderItems())
  104.                 )
  105.         );
  106.         log.info("create order success: [{}], [{}]",
  107.                 AccessContext.getLoginUserInfo().getId(), newOrder.getId());
  108.  
  109.         // 3. 扣减商品库存
  110.         if (
  111.                 !notSecuredGoodsClient.deductGoodsInventory(
  112.                         orderInfo.getOrderItems()
  113.                                 .stream()
  114.                                 .map(OrderInfo.OrderItem::toDeductGoodsInventory)
  115.                                 .collect(Collectors.toList())
  116.                 ).getData()
  117.         ) {
  118.             throw new RuntimeException("deduct goods inventory failure");
  119.         }
  120.  
  121.         // 4. 扣减用户账户余额
  122.         // 4.1 获取商品信息, 计算总价格
  123.         List<SimpleGoodsInfo> goodsInfos = notSecuredGoodsClient.getSimpleGoodsInfoByTableId(
  124.                 new TableId(
  125.                         orderInfo.getOrderItems()
  126.                                 .stream()
  127.                                 .map(o -> new TableId.Id(o.getGoodsId()))
  128.                         .collect(Collectors.toList())
  129.                 )
  130.         ).getData();
  131.         Map<Long, SimpleGoodsInfo> goodsId2GoodsInfo = goodsInfos.stream()
  132.                 .collect(Collectors.toMap(SimpleGoodsInfo::getId, Function.identity()));
  133.         long balance = 0;
  134.         for (OrderInfo.OrderItem orderItem : orderInfo.getOrderItems()) {
  135.             balance += goodsId2GoodsInfo.get(orderItem.getGoodsId()).getPrice()
  136.                     * orderItem.getCount();
  137.         }
  138.         assert balance > 0;
  139.  
  140.         // 4.2 填写总价格, 扣减账户余额
  141.         BalanceInfo balanceInfo = notSecuredBalanceClient.deductBalance(
  142.                 new BalanceInfo(AccessContext.getLoginUserInfo().getId(), balance)
  143.         ).getData();
  144.         if (null == balanceInfo) {
  145.             throw new RuntimeException("deduct user balance failure");
  146.         }
  147.         log.info("deduct user balance: [{}], [{}]", newOrder.getId(),
  148.                 JSON.toJSONString(balanceInfo));
  149.  
  150.         // 5. 发送订单物流消息 SpringCloud Stream + Kafka
  151.         LogisticsMessage logisticsMessage = new LogisticsMessage(
  152.                 AccessContext.getLoginUserInfo().getId(),
  153.                 newOrder.getId(),
  154.                 orderInfo.getUserAddress(),
  155.                 null    // 没有备注信息
  156.         );
  157.         if (!logisticsSource.logisticsOutput().send(
  158.                 MessageBuilder.withPayload(JSON.toJSONString(logisticsMessage)).build()
  159.         )) {
  160.             throw new RuntimeException("send logistics message failure");
  161.         }
  162.         log.info("send create order message to kafka with stream: [{}]",
  163.                 JSON.toJSONString(logisticsMessage));
  164.  
  165.         // 返回订单 id
  166.         return new TableId(Collections.singletonList(new TableId.Id(newOrder.getId())));
  167.     }
  168.  
  169.     @Override
  170.     public PageSimpleOrderDetail getSimpleOrderDetailByPage(int page) {
  171.  
  172.         if (page <= 0) {
  173.             page = 1;   // 默认是第一页
  174.         }
  175.  
  176.         // 这里分页的规则是: 1页10条数据, 按照 id 倒序排列
  177.         Pageable pageable = PageRequest.of(page - 1, 10,
  178.                 Sort.by("id").descending());
  179.         Page<EcommerceOrder> orderPage = orderDao.findAllByUserId(
  180.                 AccessContext.getLoginUserInfo().getId(), pageable
  181.         );
  182.         List<EcommerceOrder> orders = orderPage.getContent();
  183.  
  184.         // 如果是空, 直接返回空数组
  185.         if (CollectionUtils.isEmpty(orders)) {
  186.             return new PageSimpleOrderDetail(Collections.emptyList(), false);
  187.         }
  188.  
  189.         // 获取当前订单中所有的 goodsId, 这个 set 不可能为空或者是 null, 否则, 代码一定有 bug
  190.         Set<Long> goodsIdsInOrders = new HashSet<>();
  191.         orders.forEach(o -> {
  192.             List<DeductGoodsInventory> goodsAndCount = JSON.parseArray(
  193.                     o.getOrderDetail(), DeductGoodsInventory.class
  194.             );
  195.             goodsIdsInOrders.addAll(goodsAndCount.stream()
  196.                     .map(DeductGoodsInventory::getGoodsId)
  197.                     .collect(Collectors.toSet()));
  198.         });
  199.  
  200.         assert CollectionUtils.isNotEmpty(goodsIdsInOrders);
  201.  
  202.         // 是否还有更多页: 总页数是否大于当前给定的页
  203.         boolean hasMore = orderPage.getTotalPages() > page;
  204.  
  205.         // 获取商品信息
  206.         List<SimpleGoodsInfo> goodsInfos = securedGoodsClient.getSimpleGoodsInfoByTableId(
  207.                 new TableId(goodsIdsInOrders.stream()
  208.                         .map(TableId.Id::new).collect(Collectors.toList()))
  209.         ).getData();
  210.  
  211.         // 获取地址信息
  212.         AddressInfo addressInfo = addressClient.getAddressInfoByTablesId(
  213.                 new TableId(orders.stream()
  214.                         .map(o -> new TableId.Id(o.getAddressId()))
  215.                         .distinct().collect(Collectors.toList()))
  216.         ).getData();
  217.  
  218.         // 组装订单中的商品, 地址信息 -> 订单信息
  219.         return new PageSimpleOrderDetail(
  220.                 assembleSimpleOrderDetail(orders, goodsInfos, addressInfo),
  221.                 hasMore
  222.         );
  223.     }
  224.  
  225.     /**
  226.      * <h2>组装订单详情</h2>
  227.      * */
  228.     private List<PageSimpleOrderDetail.SingleOrderItem> assembleSimpleOrderDetail(
  229.             List<EcommerceOrder> orders, List<SimpleGoodsInfo> goodsInfos,
  230.             AddressInfo addressInfo
  231.     ) {
  232.         // goodsId -> SimpleGoodsInfo
  233.         Map<Long, SimpleGoodsInfo> id2GoodsInfo = goodsInfos.stream()
  234.                 .collect(Collectors.toMap(SimpleGoodsInfo::getId, Function.identity()));
  235.         // addressId -> AddressInfo.AddressItem
  236.         Map<Long, AddressInfo.AddressItem> id2AddressItem = addressInfo.getAddressItems()
  237.                 .stream().collect(
  238.                         Collectors.toMap(AddressInfo.AddressItem::getId, Function.identity())
  239.                 );
  240.  
  241.         List<PageSimpleOrderDetail.SingleOrderItem> result = new ArrayList<>(orders.size());
  242.         orders.forEach(o -> {
  243.  
  244.             PageSimpleOrderDetail.SingleOrderItem orderItem =
  245.                     new PageSimpleOrderDetail.SingleOrderItem();
  246.             orderItem.setId(o.getId());
  247.             orderItem.setUserAddress(id2AddressItem.getOrDefault(o.getAddressId(),
  248.                     new AddressInfo.AddressItem(-1L)).toUserAddress());
  249.             orderItem.setGoodsItems(buildOrderGoodsItem(o, id2GoodsInfo));
  250.  
  251.             result.add(orderItem);
  252.         });
  253.  
  254.         return result;
  255.     }
  256.  
  257.     /**
  258.      * <h2>构造订单中的商品信息</h2>
  259.      * */
  260.     private List<PageSimpleOrderDetail.SingleOrderGoodsItem> buildOrderGoodsItem(
  261.         EcommerceOrder order, Map<Long, SimpleGoodsInfo> id2GoodsInfo
  262.     ) {
  263.  
  264.         List<PageSimpleOrderDetail.SingleOrderGoodsItem> goodsItems = new ArrayList<>();
  265.         List<DeductGoodsInventory> goodsAndCount = JSON.parseArray(
  266.                 order.getOrderDetail(), DeductGoodsInventory.class
  267.         );
  268.  
  269.         goodsAndCount.forEach(gc -> {
  270.  
  271.             PageSimpleOrderDetail.SingleOrderGoodsItem goodsItem =
  272.                     new PageSimpleOrderDetail.SingleOrderGoodsItem();
  273.             goodsItem.setCount(gc.getCount());
  274.             goodsItem.setSimpleGoodsInfo(id2GoodsInfo.getOrDefault(gc.getGoodsId(),
  275.                     new SimpleGoodsInfo(-1L)));
  276.  
  277.             goodsItems.add(goodsItem);
  278.         });
  279.  
  280.         return goodsItems;
  281.     }
  282. }

物流微服务

创建物流微服务

引入依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>e-commerce-service</artifactId>
  7.         <groupId>com.taluohui.ecommerce</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  11.  
  12.     <artifactId>e-commerce-logistics-service</artifactId>
  13.     <version>1.0-SNAPSHOT</version>
  14.     <packaging>jar</packaging>
  15.  
  16.     <!-- 模块名及描述信息 -->
  17.     <name>e-commerce-logistics-service</name>
  18.     <description>物流服务</description>
  19.  
  20.     <properties>
  21.         <maven.compiler.source>8</maven.compiler.source>
  22.         <maven.compiler.target>8</maven.compiler.target>
  23.     </properties>
  24.  
  25.     <dependencies>
  26.         <!-- spring cloud alibaba nacos discovery 依赖 -->
  27.         <dependency>
  28.             <groupId>com.alibaba.cloud</groupId>
  29.             <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
  30.         </dependency>
  31.         <!-- zipkin = spring-cloud-starter-sleuth + spring-cloud-sleuth-zipkin-->
  32.         <dependency>
  33.             <groupId>org.springframework.cloud</groupId>
  34.             <artifactId>spring-cloud-starter-zipkin</artifactId>
  35.         </dependency>
  36.         <dependency>
  37.             <groupId>org.springframework.kafka</groupId>
  38.             <artifactId>spring-kafka</artifactId>
  39.             <version>2.5.0.RELEASE</version>
  40.         </dependency>
  41.         <!-- Java Persistence API, ORM 规范 -->
  42.         <dependency>
  43.             <groupId>org.springframework.boot</groupId>
  44.             <artifactId>spring-boot-starter-data-jpa</artifactId>
  45.         </dependency>
  46.         <!-- SpringCloud Stream + Kafka -->
  47.         <dependency>
  48.             <groupId>org.springframework.cloud</groupId>
  49.             <artifactId>spring-cloud-stream</artifactId>
  50.         </dependency>
  51.         <dependency>
  52.             <groupId>org.springframework.cloud</groupId>
  53.             <artifactId>spring-cloud-stream-binder-kafka</artifactId>
  54.         </dependency>
  55.         <!-- MySQL 驱动, 注意, 这个需要与 MySQL 版本对应 -->
  56.         <dependency>
  57.             <groupId>mysql</groupId>
  58.             <artifactId>mysql-connector-java</artifactId>
  59.             <version>8.0.12</version>
  60.             <scope>runtime</scope>
  61.         </dependency>
  62.         <dependency>
  63.             <groupId>com.taluohui.ecommerce</groupId>
  64.             <artifactId>e-commerce-service-config</artifactId>
  65.             <version>1.0-SNAPSHOT</version>
  66.         </dependency>
  67.         <dependency>
  68.             <groupId>com.taluohui.ecommerce</groupId>
  69.             <artifactId>e-commerce-service-sdk</artifactId>
  70.             <version>1.0-SNAPSHOT</version>
  71.         </dependency>
  72.     </dependencies>
  73.  
  74.     <!--
  75.         SpringBoot的Maven插件, 能够以Maven的方式为应用提供SpringBoot的支持，可以将
  76.         SpringBoot应用打包为可执行的jar或war文件, 然后以通常的方式运行SpringBoot应用
  77.      -->
  78.     <build>
  79.         <finalName>${artifactId}</finalName>
  80.         <plugins>
  81.             <plugin>
  82.                 <groupId>org.springframework.boot</groupId>
  83.                 <artifactId>spring-boot-maven-plugin</artifactId>
  84.                 <executions>
  85.                     <execution>
  86.                         <goals>
  87.                             <goal>repackage</goal>
  88.                         </goals>
  89.                     </execution>
  90.                 </executions>
  91.             </plugin>
  92.         </plugins>
  93.     </build>
  94. </project>

添加配置

  1. server:
  2.   port: 8004
  3.  
  4. spring:
  5.   main:
  6.     allow-bean-definition-overriding: true #允许同名的Bean
  7.   application:
  8.     name: e-commerce-order-service # 应用名称也是构成 Nacos 配置管理 dataId 字段的一部分 (当 config.prefix 为空时)
  9.   cloud:
  10.     stream:
  11.       kafka:
  12.         binder:
  13.           brokers: 1.15.247.9:9092
  14.           auto-create-topics: true
  15.       bindings:
  16.         logisticsInput:
  17.           destination: e-commerce-topic   # kafka topic
  18.           content-type: text/plain
  19.     alibaba:
  20.       seata:
  21.         tx-service-group: imooc-ecommerce # seata 全局事务分组
  22.     nacos:
  23.       # 服务注册发现
  24.       discovery:
  25.         enabled: true # 如果不想使用 Nacos 进行服务注册和发现, 设置为 false 即可
  26.         server-addr: 1.15.247.9:8848
  27.         # server-addr: 127.0.0.1:8848,127.0.0.1:8849,127.0.0.1:8850 # Nacos 服务器地址
  28.         namespace: 22d40198-8462-499d-a7fe-dbb2da958648
  29.   # 引入 sleuth + zipkin + kafka
  30.   kafka:
  31.     bootstrap-servers: 1.15.247.9:9092
  32.     producer:
  33.       retries: 3
  34.     consumer:
  35.       auto-offset-reset: latest
  36.   sleuth:
  37.     sampler:
  38.       # ProbabilityBasedSampler 抽样策略
  39.       probability: 1.0  # 采样比例, 1.0 表示 100%, 默认是 0.1
  40.       # RateLimitingSampler 抽样策略, 设置了限速采集, spring.sleuth.sampler.probability 属性值无效
  41.   zipkin:
  42.     sender:
  43.       type: kafka # 默认是 web
  44.     base-url: http://1.15.247.9:9411/
  45.   jpa:
  46.     show-sql: true
  47.     hibernate:
  48.       ddl-auto: none
  49.     properties:
  50.       hibernate.show_sql: true
  51.       hibernate.format_sql: true
  52.     open-in-view: false
  53.   datasource:
  54.     # 数据源
  55.     url: jdbc:mysql://1.15.247.9:3306/ecommerce?autoReconnect=true&useUnicode=true&characterEncoding=utf8&useSSL=false
  56.     username: root
  57.     password: Cjw970404
  58.     type: com.zaxxer.hikari.HikariDataSource
  59.     driver-class-name: com.mysql.cj.jdbc.Driver
  60.     # 连接池
  61.     hikari:
  62.       maximum-pool-size: 8
  63.       minimum-idle: 4
  64.       idle-timeout: 30000
  65.       connection-timeout: 30000
  66.       max-lifetime: 45000
  67.       auto-commit: true
  68.       pool-name: ImoocEcommerceHikariCP
  69.  
  70. # 暴露端点
  71. management:
  72.   endpoints:
  73.     web:
  74.       exposure:
  75.         include: '*'
  76.   endpoint:
  77.     health:
  78.       show-details: always

入口程序

  1. /**
  2.  * <h1>物流微服务启动入口</h1>
  3.  * */
  4. @Import(DataSourceProxyAutoConfiguration.class)
  5. @EnableJpaAuditing
  6. @EnableDiscoveryClient
  7. @SpringBootApplication
  8. public class LogisticsApplication {
  9.  
  10.     public static void main(String[] args) {
  11.  
  12.         SpringApplication.run(LogisticsApplication.class, args);
  13.     }
  14. }

SQL

  1. -- 创建 t_ecommerce_logistics 数据表
  2. CREATE TABLE IF NOT EXISTS `ecommerce`.`t_ecommerce_logistics` (
  3.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增主键',
  4.   `user_id` bigint(20) NOT NULL DEFAULT 0 COMMENT '用户 id',
  5.   `order_id` bigint(20) NOT NULL DEFAULT 0 COMMENT '订单 id',
  6.   `address_id` bigint(20) NOT NULL DEFAULT 0 COMMENT '用户地址记录 id',
  7.   `extra_info` varchar(512) NOT NULL COMMENT '备注信息(json 存储)',
  8.   `create_time` datetime NOT NULL DEFAULT '0000-01-01 00:00:00' COMMENT '创建时间',
  9.   `update_time` datetime NOT NULL DEFAULT '0000-01-01 00:00:00' COMMENT '更新时间',
  10.   PRIMARY KEY (`id`)
  11. ) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=utf8 COMMENT='物流表';

引入seata的两个配置文件，同上

实体类

  1. /**
  2.  * <h1>物流表实体类定义</h1>
  3.  * */
  4. @Data
  5. @NoArgsConstructor
  6. @AllArgsConstructor
  7. @Entity
  8. @EntityListeners(AuditingEntityListener.class)
  9. @Table(name = "t_ecommerce_logistics")
  10. public class EcommerceLogistics {
  11.     /** 自增主键 */
  12.     @Id
  13.     @GeneratedValue(strategy = GenerationType.IDENTITY)
  14.     @Column(name = "id", nullable = false)
  15.     private Long id;
  16.  
  17.     /** 用户 id */
  18.     @Column(name = "user_id", nullable = false)
  19.     private Long userId;
  20.  
  21.     /** 订单 id */
  22.     @Column(name = "order_id", nullable = false)
  23.     private Long orderId;
  24.  
  25.     /** 用户地址 id */
  26.     @Column(name = "address_id", nullable = false)
  27.     private Long addressId;
  28.  
  29.     /** 备注信息(json 存储) */
  30.     @Column(name = "extra_info", nullable = false)
  31.     private String extraInfo;
  32.  
  33.     /** 创建时间 */
  34.     @CreatedDate
  35.     @Column(name = "create_time", nullable = false)
  36.     private Date createTime;
  37.  
  38.     /** 更新时间 */
  39.     @LastModifiedDate
  40.     @Column(name = "update_time", nullable = false)
  41.     private Date updateTime;
  42.  
  43.     public EcommerceLogistics(Long userId, Long orderId, Long addressId, String extraInfo) {
  44.  
  45.         this.userId = userId;
  46.         this.orderId = orderId;
  47.         this.addressId = addressId;
  48.         this.extraInfo = StringUtils.isNotBlank(extraInfo) ? extraInfo : "{}";
  49.     }
  50. }

Dao

  1. /**
  2.  * <h1>EcommerceLogistics Dao 接口定义</h1>
  3.  * */
  4. public interface EcommerceLogisticsDao extends JpaRepository<EcommerceLogistics, Long> {
  5. }

service

  1. /**
  2.  * <h1>物流服务实现</h1>
  3.  * */
  4. @Slf4j
  5. @EnableBinding(LogisticsSink.class)
  6. public class LogisticsServiceImpl {
  7.  
  8.     private final EcommerceLogisticsDao logisticsDao;
  9.  
  10.     public LogisticsServiceImpl(EcommerceLogisticsDao logisticsDao) {
  11.         this.logisticsDao = logisticsDao;
  12.     }
  13.  
  14.     /**
  15.      * <h2>订阅监听订单微服务发送的物流消息</h2>
  16.      * */
  17.     @StreamListener("logisticsInput")
  18.     public void consumeLogisticsMessage(@Payload Object payload) {
  19.  
  20.         log.info("receive and consume logistics message: [{}]", payload.toString());
  21.         LogisticsMessage logisticsMessage = JSON.parseObject(
  22.                 payload.toString(), LogisticsMessage.class
  23.         );
  24.         EcommerceLogistics ecommerceLogistics = logisticsDao.save(
  25.                 new EcommerceLogistics(
  26.                         logisticsMessage.getUserId(),
  27.                         logisticsMessage.getOrderId(),
  28.                         logisticsMessage.getAddressId(),
  29.                         logisticsMessage.getExtraInfo()
  30.                 )
  31.         );
  32.         log.info("consume logistics message success: [{}]", ecommerceLogistics.getId());
  33.     }
  34. }

自定义消息通道

  1. /**
  2.  * <h1>物流微服务启动入口</h1>
  3.  * */
  4. @Import(DataSourceProxyAutoConfiguration.class)
  5. @EnableJpaAuditing
  6. @EnableDiscoveryClient
  7. @SpringBootApplication
  8. public class LogisticsApplication {
  9.  
  10.     public static void main(String[] args) {
  11.  
  12.         SpringApplication.run(LogisticsApplication.class, args);
  13.     }
  14. }
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/165667
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。