RabbitMQ+redis+Redisson分布式锁+seata实现订单服务的流程分析

public orderconfirmvo confirmorder(long skuid) {
long memberid = securityutils.getmemberid();
// 解决子线程无法获取httpservletrequest请求对象中数据的问题,子线程指getorderitemsfuture，getmemberaddressfuture，generateordertokenfuture
//feign远程调用会被拦截提取attributes并转发
requestattributes attributes = requestcontextholder.getrequestattributes();
requestcontextholder.setrequestattributes(attributes, true);
// 获取订单商品
//使用了 completablefuture 来实现异步执行获取订单项的操作
completablefuture<list<orderitemdto>> getorderitemsfuture = completablefuture.supplyasync(
	() -> this.getorderitems(skuid, memberid), threadpoolexecutor)
.exceptionally(ex -> {
	log.error("failed to get order items: {}", ex.tostring());
	return null;
});
// 用户收货地址
completablefuture<list<memberaddressdto>> getmemberaddressfuture = completablefuture.supplyasync(() -> {
	result<list<memberaddressdto>> getmemberaddressresult = memberfeignclient.listmemberaddresses(memberid);
	if (result.issuccess(getmemberaddressresult)) {
		return getmemberaddressresult.getdata();
	}
	return null;
}, threadpoolexecutor).exceptionally(ex -> {
	log.error("failed to get addresses for memberid {} : {}", memberid, ex.tostring());
	return null;
});
// 生成唯一令牌,防止重复提交(原理：提交会消耗令牌，令牌被消耗无法再次提交)
completablefuture<string> generateordertokenfuture = completablefuture.supplyasync(() -> {
	string ordertoken = this.generatetradeno(memberid);
	redistemplate.opsforvalue().set(orderconstants.order_token_prefix + ordertoken, ordertoken);
	return ordertoken;
}, threadpoolexecutor).exceptionally(ex -> {
	log.error("failed to generate order token .");
	return null;
});
//completablefuture.allof 方法，可以等待所有 completablefuture 对象都完成再进行后续操作，
// 确保获取和设置属性的操作都能够成功执行。这样可以避免程序出现异常，
completablefuture.allof(getorderitemsfuture, getmemberaddressfuture, generateordertokenfuture).join();
orderconfirmvo orderconfirmvo = new orderconfirmvo();
orderconfirmvo.setorderitems(getorderitemsfuture.join());
orderconfirmvo.setaddresses(getmemberaddressfuture.join());
orderconfirmvo.setordertoken(generateordertokenfuture.join());
log.info("order confirm response for skuid {}: {}", skuid, orderconfirmvo);
return orderconfirmvo;
}

  private string generatetradeno(long memberid) {
  //当 memberid 的位数小于 5 位时，使用 0 来填充位数不足的部分，如果 memberid 已经是 5 位数或更长，则不进行填充
  string useridfilledzero = string.format("%05d", memberid);
  //超出五位的保留后五位
  string fivedigitsuserid = useridfilledzero.substring(useridfilledzero.length() - 5);
  // 在前面加上wxo（wx order）等前缀是为了人工可以快速分辨订单号是下单还是退款、来自哪家支付机构等
  // 将时间戳+3位随机数+五位id组成商户订单号，规则参考自<a href="https://tech.meituan.com/2016/11/18/dianping-order-db-sharding.html" rel="external nofollow" >大众点评</a>
  return system.currenttimemillis() + randomutil.randomnumbers(3) + fivedigitsuserid;
  }

  /**
   * 自定义订单线程池
   *
   */
  @configuration
  @slf4j
  public class threadpoolconfig {
  	@bean
  	public threadpoolexecutor threadpoolexecutor() {
  		int cpucoresize = runtime.getruntime().availableprocessors();//使用 java 获取可用的 cpu 核心数
  		log.info("当前cpu核心数:{}", cpucoresize);
  		/*
  		 * 计算密集型: 核心线程数=cpu核心 +1   √
  		 * i/o密集型: 核心线程数=2*cpu核心 +1
  		 */
  		int corepoolsize = cpucoresize + 1;
  		return new threadpoolexecutor(
  			 //核心线程数
  				corepoolsize,
  				//最大线程数
  				2 * corepoolsize,
  				//线程空闲(存活)时间;当线程数超过核心线程数，并且空闲时间超过指定时间时，多余的线程会被销毁
  				30,
  				timeunit.seconds,
  				//任务队列(选取数组阻塞队列）特点：固定容量，公平性，先进先出
  				new arrayblockingqueue<>(1000),
  				//线程工厂
  				new namedthreadfactory("order") // 订单线程
  		);
  	}
  }

@globaltransactional
public string submitorder(ordersubmitform submitform) {
log.info("订单提交参数:{}", jsonutil.tojsonstr(submitform));
string ordertoken = submitform.getordertoken();
// 1. 判断订单是否重复提交(lua脚本保证获取和删除的原子性，成功返回1，否则返回0)
//keys[1]指orderconstants.order_token_prefix + ordertoken，argv[1]指ordertoken
string lockacquirescript = "if redis.call('get', keys[1]) == argv[1] then return redis.call('del', keys[1]) else return 0 end";
//这一行代码使用redistemplate的execute方法执行lua脚本。通过new defaultredisscript<>创建一个redis脚本对象，并指定脚本字符串和返回结果的类型（long）。
// 然后使用collections.singletonlist来创建包含锁键名和订单令牌参数的列表，传递给execute方法。
//执行 redis 脚本后，结果会赋值给 lockacquired 变量，它的类型是 long。
// lockacquired这个值表示获取锁的结果，如果成功获取并删除了锁，则为 1；如果获取锁失败，则为 0。
long lockacquired = this.redistemplate.execute(
	new defaultredisscript<>(lockacquirescript, long.class),
	collections.singletonlist(orderconstants.order_token_prefix + ordertoken),
ordertoken
);
assert.istrue(lockacquired != null && lockacquired.equals(1l), "订单重复提交，请刷新页面后重试");
// 2. 订单商品校验 (ps：校验进入订单确认页面到提交过程商品(价格、上架状态)变化)
list<ordersubmitform.orderitem> orderitems = submitform.getorderitems();
list<long> skuids = orderitems.stream()
.map(ordersubmitform.orderitem::getskuid)
.collect(collectors.tolist());
list<skuinfodto> skulist = skufeignclient.getskuinfolist(skuids);
for (ordersubmitform.orderitem item : orderitems) {
	skuinfodto skuinfo = skulist.stream().filter(sku -> sku.getid().equals(item.getskuid()))
	.findfirst()
	.orelse(null);
	assert.istrue(skuinfo != null, "商品({})已下架或删除");
	//如果调用对象小于被比较对象，compareto 方法返回一个负整数。
	// 如果调用对象大于被比较对象，compareto 方法返回一个正整数。
	assert.istrue(item.getprice().compareto(skuinfo.getprice()) == 0, "商品({})价格发生变动，请刷新页面", item.getskuname());
}
// 3. 校验库存并锁定库存
list<lockedskudto> lockedskulist = orderitems.stream()
.map(item -> new lockedskudto(item.getskuid(), item.getquantity(), item.getskusn()))
.collect(collectors.tolist());
boolean lockstockresult = skufeignclient.lockstock(ordertoken, lockedskulist);
assert.istrue(lockstockresult, "订单提交失败：锁定商品库存失败！");
// 4. 生成订单
boolean result = this.saveorder(submitform);
log.info("order ({}) create result:{}", ordertoken, result);
return ordertoken;
}

@transactional
public boolean lockstock(string ordertoken, list<lockedskudto> lockedskulist) {
	assert.istrue(collectionutil.isnotempty(lockedskulist), "订单({})未包含任何商品", ordertoken);
	// 校验库存数量是否足够以及锁定库存
	for (lockedskudto lockedsku : lockedskulist) {
		long skuid = lockedsku.getskuid();
		//商品分布式锁缓存键前缀:productconstants.sku_lock_prefix
		//每次getlock都会返回一个独立的分布式锁对象，但它们共享一个锁资源。
		rlock lock = redissonclient.getlock(productconstants.sku_lock_prefix + skuid);  // 构建商品锁对象
		try {
			//共享一个锁资源意味着多个分布式锁对象共同使用同一个锁来实现互斥访问。
			//虽然每个分布式锁对象是独立创建的，但它们会使用相同的锁资源来进行加锁和释放锁的操作。
			lock.lock();//锁定操作
			integer quantity = lockedsku.getquantity(); // 订单的商品数量
			// 库存足够
			boolean lockresult = this.update(new lambdaupdatewrapper<pmssku>()
											 .setsql("locked_stock = locked_stock + " + quantity) // 修改锁定商品数
											 .eq(pmssku::getid, lockedsku.getskuid())
											 //通过 apply 方法添加动态 sql 条件，确保 stock 减去 locked_stock 的值大于等于给定的 quantity。
											 //使用了占位符 {0} 来引用 quantity,0表示第一个传入的值。
											 .apply("stock - locked_stock >= {0}", quantity) // 剩余商品数 ≥ 订单商品数
											);
			assert.istrue(lockresult, "商品({})库存不足", lockedsku.getskusn());
		} finally {
			if (lock.islocked()) {
				lock.unlock();
			}
		}
	}
	// 锁定的商品缓存至 redis (后续使用：1.取消订单解锁库存；2：支付订单扣减库存)
	redistemplate.opsforvalue().set(productconstants.locked_skus_prefix + ordertoken, lockedskulist);
	return true;
}

/**
*解锁库存
*<p>
*订单超时未支付，释放锁定的商品库存
*/
public boolean unlockstock(string ordersn) {
//锁定库存对象集合：lockedskus
	list<lockedskudto> lockedskus = (list<lockedskudto>) redistemplate.opsforvalue().get(productconstants.locked_skus_prefix + ordersn);
	log.info("释放订单({})锁定的商品库存:{}", ordersn, jsonutil.tojsonstr(lockedskus));
	// 库存已释放
	if (collectionutil.isempty(lockedskus)) {
		return true;
	}
	// 遍历恢复锁定的商品库存
	for (lockedskudto lockedsku : lockedskus) {
		rlock lock = redissonclient.getlock(productconstants.sku_lock_prefix + lockedsku.getskuid());  // 获取商品分布式锁
		try {
			lock.lock();
			this.update(new lambdaupdatewrapper<pmssku>()
						.setsql("locked_stock = locked_stock - " + lockedsku.getquantity())
						.eq(pmssku::getid, lockedsku.getskuid())
					   );
		} finally {
			//判断当前分布式锁是否已被锁定，通过这个判断可以防止非法释放锁等潜在问题
			if (lock.islocked()) {
				lock.unlock();
			}
		}
	}
	// 移除 redis 订单锁定的商品
	redistemplate.delete(productconstants.locked_skus_prefix + ordersn);
	return true;
}

   /**
	 * 扣减库存
	 * <p>
	 * 订单支付扣减商品库存和释放锁定库存
	 *
	 * @param ordersn 订单编号
	 * @return ture/false
	 */
public boolean deductstock(string ordersn) {
	// 获取订单提交时锁定的商品
	list<lockedskudto> lockedskus = (list<lockedskudto>) redistemplate.opsforvalue().get(productconstants.locked_skus_prefix + ordersn);
	log.info("订单({})支付成功，扣减订单商品库存：{}", ordersn, jsonutil.tojsonstr(lockedskus));
	assert.istrue(collectionutil.isnotempty(lockedskus), "扣减商品库存失败：订单({})未包含商品");
	for (lockedskudto lockedsku : lockedskus) {
		rlock lock = redissonclient.getlock(productconstants.sku_lock_prefix + lockedsku.getskuid());    // 获取商品分布式锁
		try {
			lock.lock();
			this.update(new lambdaupdatewrapper<pmssku>()
					.setsql("stock = stock - " + lockedsku.getquantity())
					.setsql("locked_stock = locked_stock - " + lockedsku.getquantity())
					.eq(pmssku::getid, lockedsku.getskuid())
			);
		} finally {
			if (lock.islocked()) {
				lock.unlock();
			}
		}
	}
	// 移除订单锁定的商品
	redistemplate.delete(productconstants.locked_skus_prefix + ordersn);
	return true;
}

/**
 * 创建订单
 *
 * @param submitform 订单提交表单对象
 * @return
 */
private boolean saveorder(ordersubmitform submitform) {
//创建订单详情表(omsorder)对象
omsorder order = orderconverter.form2entity(submitform);
//设置待支付状态
order.setstatus(orderstatusenum.unpaid.getvalue());
//设置订单会员id
order.setmemberid(securityutils.getmemberid());
//设置订单来源(0代表pc订单，1代表app订单)
order.setsource(submitform.getordersource().getvalue());
//保存到数据库
boolean result = this.save(order);
long orderid = order.getid();
if (result) {
	// 保存订单明细
	list<omsorderitem> orderitementities = orderitemconverter.item2entity(submitform.getorderitems());
	orderitementities.foreach(item -> item.setorderid(orderid));
	orderitemservice.savebatch(orderitementities);
	// 订单超时未支付取消
	//这行代码使用 rabbitmq 的 java 客户端库来发送一条消息到 order.exchange 交换器，
	// 该消息会被路由到 order.close.delay 队列中。消息的内容是 submitform.getordertoken() 方法的返回结果。
	rabbittemplate.convertandsend("order.exchange", "order.close.delay", submitform.getordertoken());
}
return result;
}

@component
@slf4j
public class orderrabbitconfig {
	// 普通延迟队列
	private static final string order_close_delay_queue = "order.close.delay.queue";
	private static final string order_exchange = "order.exchange";
	private static final string order_close_delay_routing_key = "order.close.delay";
	// 死信关单队列
	private static final string order_close_queue = "order.close.queue";
	private static final string order_dlx_exchange = "order.dlx.exchange";
	private static final string order_close_routing_key = "order.close";
	/**
	 * 定义交换机
	 */
	@bean
	public exchange orderexchange() {
		return new directexchange(order_exchange, true, false);
	}
	/**
	 * 死信交换机
	 */
	@bean
	public exchange orderdlxexchange() {
		return new directexchange(order_dlx_exchange, true, false);
	}
	/**
	 * 延时队列
	 */
	@bean
	public queue orderdelayqueue() {
		// 延时队列的消息过期了，会自动触发消息的转发，根据routingkey发送到指定的exchange中，exchange路由到死信队列
		map<string, object> args = new hashmap<>();
		args.put("x-dead-letter-exchange", order_dlx_exchange);
		args.put("x-dead-letter-routing-key", order_close_routing_key); // 死信路由key
		args.put("x-message-ttl", 10 * 1000l); // 单位毫秒，10s用于测试
		return new queue(order_close_delay_queue, true, false, false, args);
	}
	/**
	 * 延时队列绑定交换机
	 */
	@bean
	public binding orderdelayqueuebinding() {
		return new binding(order_close_delay_queue, binding.destinationtype.queue, order_exchange,
						   order_close_delay_routing_key, null);
	}
	/**
	 * 关单队列
	 */
	@bean
	public queue orderclosequeue() {
		log.info("死信队列(order.close.queue)创建");
		return new queue(order_close_queue, true, false, false);
	}
	/**
	 * 关单队列绑定死信交换机
	 */
	@bean
	public binding orderclosequeuebinding() {
		return new binding(order_close_queue, binding.destinationtype.queue, order_dlx_exchange,
						   order_close_routing_key, null);
	}
}

/**
 * 订单超时未支付系统自动取消监听器
 *
 */
@component
@requiredargsconstructor
@slf4j
public class ordercloselistener {
	private final orderservice orderservice;
	private final rabbittemplate rabbittemplate;
	@rabbitlistener(queues = "order.close.queue")
	public void closeorder(string ordersn, message message, channel channel) {
		long deliverytag = message.getmessageproperties().getdeliverytag(); // 消息序号（消息队列中的位置）
		log.info("订单({})超时未支付，系统自动关闭订单", ordersn);
		try {
			boolean closeorderresult = orderservice.closeorder(ordersn);
			log.info("关单结果：{}", closeorderresult);
			if (closeorderresult) {
				// 关单成功：释放库存
				//发送订单号
				rabbittemplate.convertandsend("stock.exchange", "stock.unlock", ordersn);
			} else {
				// 关单失败：订单已被关闭，手动ack确认并从队列移除消息
				channel.basicack(deliverytag, false); // false: 不批量确认，仅确认当前单个消息
			}
		} catch (exception e) {
			// 关单异常：拒绝消息并重新入队
			try {
				channel.basicreject(deliverytag, true); //  true: 重新放回队列
				// channel.basicreject(deliverytag, false); // false: 直接丢弃消息 (todo 定时任务补偿)
			} catch (ioexception ex) {
				log.error("订单({})关闭失败，原因：{}", ordersn, ex.getmessage());
			}
		}
	}
}

  public boolean closeorder(string ordersn) {
  return this.update(new lambdaupdatewrapper<omsorder>()
  				   .eq(omsorder::getordersn, ordersn)
  				   .eq(omsorder::getstatus, orderstatusenum.unpaid.getvalue())
  				   .set(omsorder::getstatus, orderstatusenum.canceled.getvalue())
  				  );
  }

  /**
  *库存释放监听器
  */
  @component
  @slf4j
  @requiredargsconstructor
  public class stockreleaselistener {
  	private final skuservice skuservice;
  	private static final string stock_unlock_queue = "stock.unlock.queue";
  	private static final string stock_exchange = "stock.exchange";
  	private static final string stock_unlock_routing_key = "stock.unlock";
  	@rabbitlistener(bindings =
  					@queuebinding(
  						value = @queue(value = stock_unlock_queue,durable = "true"),
  						exchange = @exchange(value = stock_exchange),
  						key = {stock_unlock_routing_key}
  					),
  					ackmode = "manual")//手动ack
  	@rabbithandler
  	public void unlockstock(string ordersn, message message, channel channel){
  		log.info("订单{}取消释放库存",ordersn);
  		long delivertag=message.getmessageproperties().getdeliverytag();
  		try{
  			skuservice.unlockstock(ordersn);
  			channel.basicack(delivertag,false);
  		}catch (exception e){
  			try {
  				channel.basicack(delivertag,true);
  			}catch (ioexception ex){
  				log.error("订单{}关闭失败，原因：{}",ordersn,ex.getmessage());
  			}
  		}
  	}
  }

@globaltransactional
public <t> t payorder(orderpaymentform paymentform) {
	string ordersn = paymentform.getordersn();
	omsorder order = this.getone(new lambdaquerywrapper<omsorder>().eq(omsorder::getordersn, ordersn));
	assert.istrue(order != null, "订单不存在");
	assert.istrue(orderstatusenum.unpaid.getvalue().equals(order.getstatus()), "订单不可支付，请检查订单状态");
	rlock lock = redissonclient.getlock(orderconstants.order_lock_prefix + order.getordersn());
	try {
		lock.lock();
		t result;
		switch (paymentform.getpaymentmethod()) {
			case wx_jsapi:
				result = (t) wxjsapipay(paymentform.getappid(), order.getordersn(), order.getpaymentamount());
				break;
			default:
				result = (t) balancepay(order);
				break;
		}
		return result;
	} finally {
		//释放锁
		if (lock.islocked()) {
			lock.unlock();
		}
	}
}