Wie Springboot Redis integriert, um das Problem des Überverkaufs zu lösen

Einfacher Code für Überverkauft

Schreiben Sie einen einfachen und normalen Überverkaufs-Logikcode, und mehrere Benutzer können gleichzeitig dieselben Daten bearbeiten, um die auftretenden Probleme zu untersuchen.

Speichern Sie eine Dateninformation in Redis, fordern Sie die entsprechende Schnittstelle an und erhalten Sie Informationen zur Produktmenge.
Wenn die Produktmengeninformation größer als 0 ist, ziehen Sie 1 ab und speichern Sie sie erneut in Redis Problem.

/**
 * Redis数据库操作，超卖问题模拟
 * @author 
 *
 */
@RestController
public class RedisController {
	
	// 引入String类型redis操作模板
	@Autowired
	private StringRedisTemplate stringRedisTemplate;
 
 
	// 测试数据设置接口
	@RequestMapping("/setStock")
	public String setStock() {
		stringRedisTemplate.opsForValue().set("stock", "100");
		return "ok";
	}
	
	// 模拟商品超卖代码
	@RequestMapping("/deductStock")
	public String deductStock() {
		// 获取Redis数据库中的商品数量
		Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
		// 减库存
		if(stock > 0) {
			int realStock = stock -1;
			stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
			System.out.println("商品扣减成功，剩余商品："+realStock);
		}else {
			System.out.println("库存不足.....");
		}
		return "end";
	}
}

Überverkauftes Problem

Im Fall eines einzelnen Servers und einer einzelnen Anwendung

Verwenden Sie im Einzelanwendungsmodus jmeter für Stresstests.

jmeter压测。

 测试结果：

每个请求相当于一个线程，当几个线程同时拿到数据时，线程A拿到库存为84，这个时候线程B也进入程序，并且抢占了CPU，访问库存为84，最后两个线程都对库存减一，导致最后修改为83，实际上多卖出去了一件

既然线程和线程之间，数据处理不一致，能否使用synchronized加锁测试？

设置synchronized

依旧还是先测试单服务器

// 模拟商品超卖代码,
	// 设置synchronized同步锁
	@RequestMapping("/deductStock1")
	public String deductStock1() {
		synchronized (this) {
			// 获取Redis数据库中的商品数量
			Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
			// 减库存
			if(stock > 0) {
				int realStock = stock -1;
				stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
				System.out.println("商品扣减成功，剩余商品："+realStock);
			}else {
				System.out.println("库存不足.....");
			}
		}
		return "end";
	}

数量100

重新压测，得到的日志信息如下所示： 

 在单机模式下，添加synchronized关键字，的确能够避免商品的超卖现象！

但是在分布式微服务中，针对该服务设置了集群，synchronized依旧还能保证数据的正确性吗？

假设多个请求，被注册中心负载均衡，每个微服务中的该处理接口，都添加有synchronized，

 依然会出现类似的超卖问题：

synchronized只是针对单一服务器的JVM进行加锁，但是分布式是很多个不同的服务器，导致两个线程或多个在不同服务器上共同对商品数量信息做了操作！

---

Redis实现分布式锁 

在Redis中存在一条命令setnx (set if not exists)

setnx key value
如果不存在key，则可以设置成功；否则设置失败。

修改处理接口，增加key

// 模拟商品超卖代码
	@RequestMapping("/deductStock2")
	public String deductStock2() {
		// 创建一个key，保存至redis
		String key = "lock";
		// setnx
		// 由于redis是一个单线程，执行命令采取“队列”形式排队！
		// 优先进入队列的命令先执行，由于是setnx，第一个执行后，其他操作执行失败。
		boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, "this is lock");
		// 当不存在key时，可以设置成功，回执true；如果存在key，则无法设置，返回false
		if (!result) {
			// 前端监测，redis中存在，则不能让这个抢购操作执行，予以提示！
			return "err";
		}
		
		// 获取Redis数据库中的商品数量
		Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
		// 减库存
		if(stock > 0) {
			int realStock = stock -1;
			stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
			System.out.println("商品扣减成功，剩余商品："+realStock);
		}else {
			System.out.println("库存不足.....");
		}
 
        // 程序执行完成，则删除这个key
		stringRedisTemplate.delete(key);
 
		return "end";
	}

1、请求进入接口中，如果redis中不存在key，则会新建一个setnx；如果存在，则不会新建，同时返回错误编码，不会继续执行抢购逻辑。
2、当创建成功后，执行抢购逻辑。
3、抢购逻辑执行完成后，删除数据库中对应的setnx的key。让其他请求能够设置并操作。

这种逻辑来说比之前单一使用syn合理的多，但是如果执行抢购操作中出现了异常，导致这个key无法被删除。以至于其他处理请求，一直无法拿到key，程序逻辑死锁！

可以采取try … finally进行操作 

/**
	 * 模拟商品超卖代码 设置
	 *
	 * @return
	 */
	@RequestMapping("/deductStock3")
	public String deductStock3() {
		// 创建一个key，保存至redis
		String key = "lock";
		// setnx
		// 由于redis是一个单线程，执行命令采取队列形式排队！优先进入队列的命令先执行，由于是setnx，第一个执行后，其他操作执行失败
		boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, "this is lock");
		// 当不存在key时，可以设置成功，回执true；如果存在key，则无法设置，返回false
		if (!result) {
			// 前端监测，redis中存在，则不能让这个抢购操作执行，予以提示！
			return "err";
		}
 
		try {
			// 获取Redis数据库中的商品数量
			Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
			// 减库存
			if (stock > 0) {
				int realStock = stock - 1;
				stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
				System.out.println("商品扣减成功，剩余商品：" + realStock);
			} else {
				System.out.println("库存不足.....");
			}
		} finally {
			// 程序执行完成，则删除这个key
			// 放置于finally中，保证即使上述逻辑出问题，也能del掉
			stringRedisTemplate.delete(key);
		}
 
		return "end";
	}

这个逻辑相比上面其他的逻辑来说，显得更加的严谨。

但是，如果一套服务器，因为断电、系统崩溃等原因出现宕机，导致本该执行finally中的语句未成功执行完成！！同样出现key一直存在，导致死锁！

通过超时间解决上述问题

在设置成功setnx后，以及抢购代码逻辑执行前，增加key的限时。

/**
	 * 模拟商品超卖代码 设置setnx保证分布式环境下，数据处理安全行问题；<br>
	 * 但如果某个代码段执行异常，导致key无法清理，出现死锁，添加try...finally;<br>
	 * 如果某个服务因某些问题导致释放key不能执行，导致死锁，此时解决思路为：增加key的有效时间;<br>
	 * 为了保证设置key的值和设置key的有效时间，两条命令构成同一条原子命令，将下列逻辑换成其他代码。
	 *
	 * @return
	 */
	@RequestMapping("/deductStock4")
	public String deductStock4() {
		// 创建一个key，保存至redis
		String key = "lock";
		// setnx
		// 由于redis是一个单线程，执行命令采取队列形式排队！优先进入队列的命令先执行，由于是setnx，第一个执行后，其他操作执行失败
		//boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, "this is lock");
 
		//让设置key和设置key的有效时间都可以同时执行
		boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, "this is lock", 10, TimeUnit.SECONDS);
 
		// 当不存在key时，可以设置成功，回执true；如果存在key，则无法设置，返回false
		if (!result) {
			// 前端监测，redis中存在，则不能让这个抢购操作执行，予以提示！
			return "err";
		}
		// 设置key有效时间
		//stringRedisTemplate.expire(key, 10, TimeUnit.SECONDS);
 
		try {
			// 获取Redis数据库中的商品数量
			Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
			// 减库存
			if (stock > 0) {
				int realStock = stock - 1;
				stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
				System.out.println("商品扣减成功，剩余商品：" + realStock);
			} else {
				System.out.println("库存不足.....");
			}
		} finally {
			// 程序执行完成，则删除这个key
			// 放置于finally中，保证即使上述逻辑出问题，也能del掉
			stringRedisTemplate.delete(key);
		}
 
		return "end";
	}

但是上述代码的逻辑中依旧会有问题：

如果处理逻辑中，出现超时

Testergebnisse:

Jede Anfrage entspricht einem Thread, Wenn mehrere Threads gleichzeitig die Daten abrufen, erhält Thread A den Bestand als 84. Zu diesem Zeitpunkt tritt auch Thread B in das Programm ein, belegt die CPU und greift auf den Bestand als 84 zu. Am Ende dekrementieren beide Threads den Bestand um eins, was zur endgültigen Änderung von 83 führte. Tatsächlich wurde ein weiterer Artikel verkauft

Kann ich synchronized verwenden, um den Test zu sperren, da die Datenverarbeitung zwischen den Threads inkonsistent ist?

Synchronisiert einstellen

🎜Testen Sie immer noch zuerst einen einzelnen Server🎜

/**
	 * 模拟商品超卖代码 <br>
	 * 解决`deductStock6`中，key形同虚设的问题。
	 *
	 * @return
	 */
	@RequestMapping("/deductStock5")
	public String deductStock5() {
		// 创建一个key，保存至redis
		String key = "lock";
		String lock_value = UUID.randomUUID().toString();
		// setnx
		//让设置key和设置key的有效时间都可以同时执行
		boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, lock_value, 10, TimeUnit.SECONDS);
		// 当不存在key时，可以设置成功，回执true；如果存在key，则无法设置，返回false
		if (!result) {
			// 前端监测，redis中存在，则不能让这个抢购操作执行，予以提示！
			return "err";
		}
		try {
			// 获取Redis数据库中的商品数量
			Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
			// 减库存
			if (stock > 0) {
				int realStock = stock - 1;
				stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
				System.out.println("商品扣减成功，剩余商品：" + realStock);
			} else {
				System.out.println("库存不足.....");
			}
		} finally {
			// 程序执行完成，则删除这个key
			// 放置于finally中，保证即使上述逻辑出问题，也能del掉
 
			// 判断redis中该数据是否是这个接口处理时的设置的，如果是则删除
			if(lock_value.equalsIgnoreCase(stringRedisTemplate.opsForValue().get(key))) {
				stringRedisTemplate.delete(key);
			}
		}
		return "end";
	}

🎜Menge100🎜🎜🎜🎜Erneuter Stresstest, die erhaltenen Protokollinformationen lauten wie folgt: 🎜🎜🎜🎜 Im Standalone-Modus kann das Hinzufügen des synchronisierten Schlüsselworts tatsächlich das Phänomen des Überverkaufs von Produkten vermeiden! 🎜🎜Aber kann in einem verteilten Microservice, wenn ein Cluster für den Service eingerichtet wird, die Synchronisierung trotzdem die Richtigkeit der Daten garantieren? 🎜🎜Angenommen, dass mehrere Anfragen vom Registrierungszentrum belastet werden, wird die Verarbeitungsschnittstelle in jedem Mikrodienst synchronisiert hinzugefügt, 🎜🎜🎜🎜 Ähnliche überverkaufte-Probleme werden weiterhin auftreten: 🎜🎜🎜synchronisiertcode> ist nur gesperrt für die JVM eines einzelnen Servers, aber auf viele verschiedene Server verteilt, was zu zwei Threads oder mehr führt. Die Produktmengeninformationen wurden gemeinsam betrieben auf verschiedenen Servern! 🎜🎜

---

🎜Redis implementiert verteilte Sperren 🎜🎜Es gibt einen Befehl in Redis setnx (set if not exist)🎜🎜🎜setnx-Schlüsselwert🎜Wenn der Schlüssel nicht existiert, kann er vorhanden sein erfolgreich gesetzt; andernfalls schlägt die Einstellung fehl. 🎜🎜🎜Ändern Sie die Verarbeitungsschnittstelle und fügen Sie einen Schlüssel hinzu🎜

@Component
public class RedisLock {
    private final Logger log = LoggerFactory.getLogger(this.getClass());
 
    private final long acquireTimeout = 10*1000;    // 获取锁之前的超时时间(获取锁的等待重试时间)
    private final int timeOut = 20;   // 获取锁之后的超时时间(防止死锁)
 
    @Autowired
    private StringRedisTemplate stringRedisTemplate;  // 引入String类型redis操作模板
 
    /**
     * 获取分布式锁
     * @return 锁标识
     */
    public boolean getRedisLock(String lockName,String lockValue) {
        // 1.计算获取锁的时间
        Long endTime = System.currentTimeMillis() + acquireTimeout;
        // 2.尝试获取锁
        while (System.currentTimeMillis() < endTime) {
            //3. 获取锁成功就设置过期时间 让设置key和设置key的有效时间都可以同时执行
            boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockName, lockValue, timeOut, TimeUnit.SECONDS);
            if (result) {
                return true;
            }
        }
        return false;
    }
 
 
    /**
     * 释放分布式锁
     * @param lockName 锁名称
     * @param lockValue 锁值
     */
    public void unRedisLock(String lockName,String lockValue) {
        if(lockValue.equalsIgnoreCase(stringRedisTemplate.opsForValue().get(lockName))) {
            stringRedisTemplate.delete(lockName);
        }
    }
}

🎜🎜1 Wenn der Schlüssel in Redis nicht vorhanden ist, wird ein neuer Satz erstellt Es wird ein Fehlercode zurückgegeben und die Ausführung der Rush-Kauflogik wird nicht fortgesetzt. 🎜2. Nachdem die Erstellung erfolgreich war, führen Sie die Snap-Up-Logik aus. 🎜3. Nachdem die Eilkauflogik ausgeführt wurde, löschen Sie den Schlüssel des entsprechenden setnx in der Datenbank. Ermöglicht die Einrichtung und Bearbeitung anderer Anfragen. 🎜🎜🎜Diese Logik ist viel sinnvoller als die alleinige Verwendung von syn. Wenn jedoch während des Eilkaufvorgangs eine Ausnahme auftritt, kann der Schlüssel nicht von gelöscht werden . Infolgedessen konnten andere Verarbeitungsanforderungen den Schlüssel nicht erhalten und die Programmlogik ist blockiert! 🎜🎜Sie können try...finally verwenden, um 🎜

@RestController
public class RedisController {
	
	// 引入String类型redis操作模板
	@Autowired
	private StringRedisTemplate stringRedisTemplate;
	@Autowired
	private RedisLock redisLock;
 
 
	@RequestMapping("/setStock")
	public String setStock() {
		stringRedisTemplate.opsForValue().set("stock", "100");
		return "ok";
	}
 
	@RequestMapping("/deductStock")
	public String deductStock() {
		// 创建一个key，保存至redis
		String key = "lock";
		String lock_value = UUID.randomUUID().toString();
		try {
			boolean redisLock = this.redisLock.getRedisLock(key, lock_value);//获取锁
			if (redisLock)
			{
				// 获取Redis数据库中的商品数量
				Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
				// 减库存
				if (stock > 0) {
					int realStock = stock - 1;
					stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
					System.out.println("商品扣减成功，剩余商品：" + realStock);
				} else {
					System.out.println("库存不足.....");
				}
			}
		} finally {
			redisLock.unRedisLock(key,lock_value);   //释放锁
		}
		return "end";
	}
}

🎜Diese Logik ist strenger als die andere Logik oben. 🎜🎜Wenn jedoch eine Reihe von Servern aufgrund von Stromausfällen, Systemabstürzen usw. ausgefallen ist, werden die Anweisungen in finally, die ausgeführt werden sollten, nicht erfolgreich ausgeführt! ! Es scheint auch, dass der Schlüssel immer existiert, was zu einem Deadlock führt! 🎜🎜Lösen Sie das obige Problem durch Timeout. 🎜🎜Erhöhen Sie nach dem erfolgreichen Festlegen von setnx und vor der Ausführung der Snap-Up-Code-Logik das Schlüsselzeitlimit. 🎜rrreee🎜Aber es wird immer noch Probleme in der Logik des obigen Codes geben: 🎜🎜🎜Wenn es ein timeout-Problem in der Verarbeitungslogik gibt. 🎜Welche Probleme treten zu diesem Zeitpunkt auf, wenn die Logik ausgeführt wird und die Zeit die festgelegte Schlüsselgültigkeitszeit überschreitet? 🎜🎜🎜🎜🎜🎜🎜 Das Problem ist auf dem obigen Bild deutlich zu erkennen: 🎜Wenn die Ausführungszeit einer Anfrage die gültige Zeit des Schlüssels überschreitet. 🎜Wenn eine neue Anfrage ausgeführt wird, können Sie auf jeden Fall den Schlüssel abrufen und die Zeit festlegen. 🎜Der zu diesem Zeitpunkt in Redis gespeicherte Schlüssel ist nicht der Schlüssel von Anfrage 1, sondern wird von anderen Anfragen festgelegt. 🎜Nachdem die Ausführung von Anfrage 1 abgeschlossen ist, wird der Schlüssel hier gelöscht. Was gelöscht wird, ist der von anderen Anfragen festgelegte Schlüssel! 🎜

依然出现了key形同虚设的问题！如果失效一直存在，超卖问题依旧不会解决。

通过key设置值匹配的方式解决形同虚设问题 

既然出现key形同虚设的现象，是否可以增加条件，当finally中需要执行删除操作时，获取数据判断值是否是该请求中对应的，如果是则删除，不是则不管！

修改上述代码如下所示：

/**
	 * 模拟商品超卖代码 <br>
	 * 解决`deductStock6`中，key形同虚设的问题。
	 *
	 * @return
	 */
	@RequestMapping("/deductStock5")
	public String deductStock5() {
		// 创建一个key，保存至redis
		String key = "lock";
		String lock_value = UUID.randomUUID().toString();
		// setnx
		//让设置key和设置key的有效时间都可以同时执行
		boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, lock_value, 10, TimeUnit.SECONDS);
		// 当不存在key时，可以设置成功，回执true；如果存在key，则无法设置，返回false
		if (!result) {
			// 前端监测，redis中存在，则不能让这个抢购操作执行，予以提示！
			return "err";
		}
		try {
			// 获取Redis数据库中的商品数量
			Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
			// 减库存
			if (stock > 0) {
				int realStock = stock - 1;
				stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
				System.out.println("商品扣减成功，剩余商品：" + realStock);
			} else {
				System.out.println("库存不足.....");
			}
		} finally {
			// 程序执行完成，则删除这个key
			// 放置于finally中，保证即使上述逻辑出问题，也能del掉
 
			// 判断redis中该数据是否是这个接口处理时的设置的，如果是则删除
			if(lock_value.equalsIgnoreCase(stringRedisTemplate.opsForValue().get(key))) {
				stringRedisTemplate.delete(key);
			}
		}
		return "end";
	}

由于获得锁的线程必须执行完减库存逻辑才能释放锁，所以在此期间所有其他的线程都会由于没获得锁，而直接结束程序，导致有很多库存根本没有卖出去，所以这里应该可以优化，让没获得锁的线程等待，或者循环检查锁 

---

最终版

我们将锁封装到一个实体类中，然后加入两个方法，加锁和解锁

@Component
public class RedisLock {
    private final Logger log = LoggerFactory.getLogger(this.getClass());
 
    private final long acquireTimeout = 10*1000;    // 获取锁之前的超时时间(获取锁的等待重试时间)
    private final int timeOut = 20;   // 获取锁之后的超时时间(防止死锁)
 
    @Autowired
    private StringRedisTemplate stringRedisTemplate;  // 引入String类型redis操作模板
 
    /**
     * 获取分布式锁
     * @return 锁标识
     */
    public boolean getRedisLock(String lockName,String lockValue) {
        // 1.计算获取锁的时间
        Long endTime = System.currentTimeMillis() + acquireTimeout;
        // 2.尝试获取锁
        while (System.currentTimeMillis() < endTime) {
            //3. 获取锁成功就设置过期时间 让设置key和设置key的有效时间都可以同时执行
            boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockName, lockValue, timeOut, TimeUnit.SECONDS);
            if (result) {
                return true;
            }
        }
        return false;
    }
 
 
    /**
     * 释放分布式锁
     * @param lockName 锁名称
     * @param lockValue 锁值
     */
    public void unRedisLock(String lockName,String lockValue) {
        if(lockValue.equalsIgnoreCase(stringRedisTemplate.opsForValue().get(lockName))) {
            stringRedisTemplate.delete(lockName);
        }
    }
}

@RestController
public class RedisController {
	
	// 引入String类型redis操作模板
	@Autowired
	private StringRedisTemplate stringRedisTemplate;
	@Autowired
	private RedisLock redisLock;
 
 
	@RequestMapping("/setStock")
	public String setStock() {
		stringRedisTemplate.opsForValue().set("stock", "100");
		return "ok";
	}
 
	@RequestMapping("/deductStock")
	public String deductStock() {
		// 创建一个key，保存至redis
		String key = "lock";
		String lock_value = UUID.randomUUID().toString();
		try {
			boolean redisLock = this.redisLock.getRedisLock(key, lock_value);//获取锁
			if (redisLock)
			{
				// 获取Redis数据库中的商品数量
				Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
				// 减库存
				if (stock > 0) {
					int realStock = stock - 1;
					stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
					System.out.println("商品扣减成功，剩余商品：" + realStock);
				} else {
					System.out.println("库存不足.....");
				}
			}
		} finally {
			redisLock.unRedisLock(key,lock_value);   //释放锁
		}
		return "end";
	}
}

可以看到失败的线程不会直接结束，而是会尝试重试，一直到重试结束时间，才会结束

---

实际上这个最终版依然存在3个问题

1、在finally流程中，由于是先判断在处理。如果判断条件结束后，获取到的结果为true。但是在执行del操作前，此时jvm在执行GC操作(为了保证GC操作获取GC roots根完全，会暂停java程序)，导致程序暂停。在GC操作完成并恢复后，执行del操作时，当前被加锁的key是否仍然存在？

2、问题如图所示

Das obige ist der detaillierte Inhalt vonWie Springboot Redis integriert, um das Problem des Überverkaufs zu lösen. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!