当前位置：首页 > news >正文

分布式事务、锁、链路追踪

news 2025/8/15 12:14:17

一、链路追踪

基本介绍

下载方式

如何使用

链路跟踪

追踪调用链

二、动态（多）数据源切换

一个项目，连接多个数据库

不同接口访问不同数据库

三、分布式事务seata

什么是分布式事务

四、分布式锁setnx|redisson

redis分布式锁实现

1、setnx实现分布式锁

2、redisson 实现分布式锁

一、链路追踪

基本介绍

什么是链路追踪

随着微服务分布式系统变得日趋复杂，越来越多的组件开始走向分布式化，如分布式服务、分布式数据库、分布式缓存等，使得后台服务构成了一种复杂的分布式网络。在服务能力提升的同时，复杂的网络结构也使问题定位更加困难。在一个请求在经过诸多服务过程中，出现了某一个调用失败的情况，查询具体的异常由哪一个服务引起的就变得十分抓狂，问题定位和处理效率是也会非常低。

分布式链路追踪就是将一次分布式请求还原成调用链路，将一次分布式请求的调用情况集中展示，比如各个服务节点上的耗时、请求具体到达哪台机器上、每个服务节点的请求状态等等。

为什么要使用链路追踪

链路追踪为分布式应用的开发者提供了完整的调用链路还原、调用请求量统计、链路拓扑、应用依赖分析等工具，可以帮助开发者快速分析和诊断分布式应用架构下的性能瓶颈，提高微服务时代下的开发诊断效率。

skywalking 链路追踪

SkyWalking 是一个可观测性分析平台（Observability Analysis Platform 简称OAP）和应用性能管理系统（Application Performance Management 简称 APM）。

提供分布式链路追踪，服务网格（Service Mesh）遥测分析，度量（Metric）聚合和可视化一体化解决方案。

SkyWalking 特点

多语言自动探针，java，.Net Code ,Node.Js
多监控手段，语言探针和Service Mesh
轻量高效，不需要额外搭建大数据平台
模块化架构，UI ，存储《集群管理多种机制可选
支持警告
优秀的可视化效果

下面是 SkyWalking 的架构图：

下载方式

Windows平台安装包下载

可以从 http://skywalking.apache.org/downloads 下载 apache-skywalking

apm-$version.tar.gz 包。

Windows下载解压后（.tar.gz），直接点击 bin/startup.bat 就可以了，这个时候实际上是启动了两

个项目，一个收集器，一个web页面。

提示

如果觉得官网下载慢，可以使用我分享的网盘地址: https://pan.baidu.com/s/1E9J52g6uW_VFWY34fHL6zA 提取码: vneh

打开控制台

skywalking 提供了一个可视化的监控平台，安装好之后，在浏览器中输入(http://localhost:8080

(opens new window))就可以访问了。（我使用的是8.3.0版本）

如何使用

配置vm参数

idea 配置 vm 参数图：

eclipse 配置 vm 参数图：

-javaagent:D:\apache-skywalking-apm-bin\agent\skywalking-agent.jar
-Dskywalking.agent.service_name=ruoyi-gateway
-Dskywalking.collector.backend_service=localhost:11800

启动项目，访问接口，再去(http://localhost:8080 (opens new window))看面板数据

参数	描述
`javaagent`	配置skywalking-agent.jar的地址
`service_name`	配置需要监控的服务名
`javaagent`	skywalking收集器服务的地址

链路跟踪

当我们访问一个服务，而他会调用另一个服务的时候，点击拓扑图会出现下图的效果，这就是链路跟踪的效果

追踪调用链

在追踪界面，可以查看整个请求的具体调用链

二、动态（多）数据源切换

一个项目，连接多个数据库

不同接口访问不同数据库

spring:datasource:dynamic:druid:initial-size: 5min-idle: 5maxActive: 20maxWait: 60000datasource:master:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://localhost:3306/ry-cloud?
useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL
=true&serverTimezone=GMT%2B8username: rootpassword: 123456order:username: rootpassword: 123456url: jdbc:mysql://localhost:3306/seata_order?
useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL
=true&serverTimezone=GMT%2B8driver-class-name: com.mysql.cj.jdbc.Driveraccount:username: rootpassword: 123456url: jdbc:mysql://localhost:3306/seata_account?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL
=true&serverTimezone=GMT%2B8driver-class-name: com.mysql.cj.jdbc.Driverproduct:username: rootpassword: 123456url: jdbc:mysql://localhost:3306/seata_product?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL
=true&serverTimezone=GMT%2B8driver-class-name: com.mysql.cj.jdbc.Driver

service方法上，通过指定@Ds（"数据源"）进行数据源切换

//省略不写 访问master
public void placeOrder(PlaceOrderRequest request)
{}
@Ds("product")
public void placeOrder(PlaceOrderRequest request)
{}
@Ds("order")
public void placeOrder(PlaceOrderRequest request)
{}

三、分布式事务seata

下单(5张表)

5张表在同一个数据库中,同一个模块中（

@Transactional

service层一个方法（）{

调用5个service|mapper方法

}

）

5张表在5个数据库中（

@GlobalTransactional

service层一个方法（）{

调用5个service方法（ @Transactional 每一个方法都有自己的数据库）

}

）

创建订单（insert into orer）

创建订单详情(insert into order_detail)

判断库存是否充足（select stock from product ）

减少库存（update product ）

修改账户余额（update account）

提醒商家发货（insert msg）

seata:enabled: true

什么是分布式事务

指一次大的操作由不同的小操作组成的，这些小的操作分布在不同的服务器上，分布式事务需要保证这些小操作要么全部成功，要么全部失败。从本质上来说，分布式事务就是为了保证不同数据库的数据一致性。

@DS("order") // 每一层都需要使用多数据源注解切换所选择的数据库
@Override
@Transactional
@GlobalTransactional // 重点 第一个开启事务的需要添加seata全局事务注解
public void placeOrder(PlaceOrderRequest request)
{log.info("=============ORDER START=================");Long userId = request.getUserId();Long productId = request.getProductId();Integer amount = request.getAmount();log.info("收到下单请求,用户:{}, 商品:{},数量:{}", userId, productId, amount);Order order = new Order(userId, productId, 0,amount);orderMapper.insert(order);log.info("订单一阶段生成，等待扣库存付款中");// 扣减库存并计算总价Double totalPrice = productService.reduceStock(productId, amount);// 扣减余额accountService.reduceBalance(userId, totalPrice);order.setStatus(1);order.setTotalPrice(totalPrice);orderMapper.updateById(order);log.info("订单已成功下单");log.info("=============ORDER END=================");
}

/**
* 事务传播特性设置为 REQUIRES_NEW 开启新的事务 重要！！！！一定要使用REQUIRES_NEW
*/@DS("product")@Transactional(propagation = Propagation.REQUIRES_NEW)@Overridepublic Double reduceStock(Long productId, Integer amount){log.info("=============PRODUCT START=================");log.info("当前 XID: {}", RootContext.getXID());// 检查库存Product product = productMapper.selectById(productId);Integer stock = product.getStock();log.info("商品编号为 {} 的库存为{},订单商品数量为{}", productId, stock,amount);if (stock < amount){log.warn("商品编号为{} 库存不足，当前库存:{}", productId, stock);throw new RuntimeException("库存不足");}log.info("开始扣减商品编号为 {} 库存,单价商品价格为{}", productId,product.getPrice());// 扣减库存int currentStock = stock - amount;product.setStock(currentStock);productMapper.updateById(product);double totalPrice = product.getPrice() * amount;log.info("扣减商品编号为 {} 库存成功,扣减后库存为{}, {} 件商品总价为 {} ",productId, currentStock, amount, totalPrice);log.info("=============PRODUCT END=================");return totalPrice;}
}

package com.ruoyi.seckill.controller;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.ruoyi.seckill.domain.dto.PlaceOrderRequest;
import com.ruoyi.seckill.service.OrderService;
import io.swagger.annotations.ApiOperation;
@RestController
@RequestMapping("/order")
public class OrderController
{@Autowiredprivate OrderService orderService;@PostMapping("/placeOrder")public String placeOrder(@Validated @RequestBody PlaceOrderRequest request){orderService.placeOrder(request);return "下单成功";}@PostMapping("/test1")@ApiOperation("测试商品库存不足-异常回滚")public String test1(){// 商品单价10元，库存20个,用户余额50元，模拟一次性购买22个。 期望异常回滚orderService.placeOrder(new PlaceOrderRequest(1L, 1L, 22));return "下单成功";}@PostMapping("/test2")@ApiOperation("测试用户账户余额不足-异常回滚")public String test2(){// 商品单价10元，库存20个，用户余额50元，模拟一次性购买6个。 期望异常回滚orderService.placeOrder(new PlaceOrderRequest(1L, 1L, 6));return "下单成功";}
}

四、分布式锁setnx|redisson

针对一段代码，要求同一时刻，只有一个线程能够访问。

javase 同步锁 synchronized Lock 应用于单体架构中。

redis 实现分布式锁

redis分布式锁实现

1、setnx实现分布式锁

setnx(key,value) == true,

设置一个key，如果key不存在，设置成功，如果key存在，设置失败。

2、redisson 实现分布式锁

可以实现公平锁和可重入。

/*
* Copyright 2013-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.ruoyi.seckill.demos.web;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.ModelAttribute;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import java.util.concurrent.TimeUnit;
/**
* @author <a href="mailto:chenxilzx1@gmail.com">theonefx</a>
*/
@Controller
public class BasicController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Autowiredprivate RedissonClient redissonClient;@RequestMapping("/unLock")@ResponseBodypublic String unLock() {try{String stock = stringRedisTemplate.opsForValue().get("stock");int stockNum = Integer.parseInt(stock);if(stockNum > 0){//设置库存减1int realStock = stockNum - 1;stringRedisTemplate.opsForValue().set("stock",realStock + "");System.out.println("设置库存" + realStock);}else{System.out.println("库存不足");}}catch(Exception e){e.printStackTrace();}return "Hello ";}@RequestMapping("/setnx")@ResponseBodypublic String setnx() {try{RedisConnection connection =stringRedisTemplate.getConnectionFactory().getConnection();if(connection.setNX("m1".getBytes(),"v1".getBytes())){//没有排队机制String stock = stringRedisTemplate.opsForValue().get("stock");int stockNum = Integer.parseInt(stock);if(stockNum > 0){//设置库存减1int realStock = stockNum - 1;stringRedisTemplate.opsForValue().set("stock",realStock +"");System.out.println("设置库存" + realStock);}else{System.out.println("库存不足");}connection.del("m1".getBytes());}}catch(Exception e){e.printStackTrace();}return "Hello ";}@RequestMapping("/hello")@ResponseBodypublic String hello() {RLock lock = redissonClient.getLock("redisson:stockLock");try{//加锁lock.lock(10, TimeUnit.SECONDS);String stock = stringRedisTemplate.opsForValue().get("stock");int stockNum = Integer.parseInt(stock);if(stockNum > 0){//设置库存减1int realStock = stockNum - 1;stringRedisTemplate.opsForValue().set("stock",realStock + "");System.out.println("设置库存" + realStock);}else{System.out.println("库存不足");}}catch(Exception e){e.printStackTrace();}finally {lock.unlock();}return "Hello ";}
}