Bagaimana Springboot mengintegrasikan Redis untuk mencapai masalah terlalu banyak jualan

Kod mudah untuk terlebih jual

Tulis kod logik terlebih jual yang mudah dan biasa Berbilang pengguna boleh mengendalikan sekeping data yang sama pada masa yang sama untuk meneroka masalah yang timbul.

Simpan sekeping maklumat data dalam Redis, minta antara muka yang sepadan dan dapatkan maklumat kuantiti produk
Jika maklumat kuantiti produk lebih daripada 0, tolak 1 dan simpan semula dalam Redis
Menjalankan masalah ujian kod .

/**
 * 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";
	}
}

Masalah terlebih jual

Dalam kes pelayan tunggal dan aplikasi tunggal

Dalam mod aplikasi tunggal, gunakan jmeter ujian tekanan.

Keputusan ujian:

Setiap permintaan adalah bersamaan dengan urutan, Apabila beberapa utas mendapat data pada masa yang sama, utas A mendapat inventori sebagai 84. Pada masa ini, utas B juga memasuki program dan merampas CPU, mengakses inventori sebagai 84. Akhirnya, kedua-dua utas mengurangkan inventori dengan satu, menghasilkan pengubahsuaian terakhir sebagai 83. Malah, satu lagi item telah dijual

Memandangkan pemprosesan data tidak konsisten antara benang, bolehkah kita menggunakan synchronized ujian mengunci?

Tetapkan disegerakkan

Masih menguji satu pelayan dahulu

// 模拟商品超卖代码,
	// 设置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";
	}

Kuantiti100

Ujian tekanan semula , Maklumat log yang diperolehi adalah seperti berikut:

Dalam mod bersendirian, menambah kata kunci yang disegerakkan sememangnya boleh mengelakkan fenomena terlebih jual produk!

Tetapi dalam perkhidmatan mikro yang diedarkan, jika kluster disediakan untuk perkhidmatan itu, bolehkah disegerakkan masih menjamin ketepatan data?

Dengan mengandaikan bahawa berbilang permintaan diimbangi beban oleh pusat pendaftaran, antara muka pemprosesan dalam setiap perkhidmatan mikro ditambah dengan disegerakkan,

masih akan muncul Serupa 超卖 masalah:

synchronized hanya untuk 单一服务器 daripada JVM, tetapi pengedaran adalah banyak pelayan yang berbeza, menghasilkan dua rangkaian atau lebih Maklumat kuantiti produk dikendalikan secara bersama pada pelayan yang berbeza ! 加锁

---

Redis melaksanakan kunci teragih

Terdapat arahan dalam Redis

setnx (set if not exists)

nilai kunci setnx

Jika tidak kekunci wujud, tetapan boleh berjaya; jika tidak tetapan gagal.

Ubah suai antara muka pemprosesan dan tambah kunci

// 模拟商品超卖代码
	@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. Permintaan memasuki antara muka Jika kunci tidak wujud dalam redis, setnx baharu akan dibuat; jika ia wujud, ia tidak akan Ia akan baru dibuat dan kod ralat akan dikembalikan pada masa yang sama, dan logik pembelian tergesa-gesa tidak akan terus dilaksanakan.

2. Selepas penciptaan berjaya, laksanakan logik snap-up.
3. Selepas logik pembelian tergesa-gesa dilaksanakan, padamkan
dan setnx yang sepadan dalam pangkalan data. Membolehkan permintaan lain disediakan dan diambil tindakan. key

Logik ini jauh lebih munasabah daripada menggunakan

sahaja sebelum ini, namun, jika pengecualian berlaku semasa operasi snap-up, ini syn tidak boleh key. Supaya permintaan pemprosesan lain tidak dapat 删除, dan logik program buntu! key

Anda boleh menggunakan try...akhirnya untuk beroperasi

/**
	 * 模拟商品超卖代码 设置
	 *
	 * @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";
	}

Logik ini lebih ketat daripada logik lain di atas.

Walau bagaimanapun, jika pelayan menghadapi

disebabkan gangguan bekalan elektrik, sistem ranap, dsb., pernyataan dalam 宕机 yang sepatutnya dilaksanakan tidak berjaya dilaksanakan! ! finally turut muncul, menghasilkan key一直存在! 死锁

Selesaikan masalah di atas melalui tamat masa

Selepas tetapan berjaya

dan sebelum logik kod pembelian tergesa-gesa dilaksanakan, tambahkan had masa kunci. setnx

/**
	 * 模拟商品超卖代码 设置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";
	}

Tetapi masih akan ada masalah dalam logik kod di atas:

Jika terdapat masalah

dalam logik pemprosesan. 超时Apabila logik dilaksanakan dan masa melebihi masa kesahan kunci yang ditetapkan, apakah masalah yang akan berlaku pada masa ini?

Masalahnya boleh dilihat dengan jelas daripada gambar di atas:

Jika masa pelaksanaan permintaan melebihi masa sah kunci.
Apabila permintaan baharu dilaksanakan, anda pasti akan dapat mendapatkan kunci dan menetapkan masa
Kunci yang disimpan dalam redis pada masa ini bukanlah kunci permintaan 1, tetapi ditetapkan oleh permintaan lain.
Selepas pelaksanaan permintaan 1 selesai, kunci dipadamkan di sini. Apa yang dipadamkan ialah kunci yang ditetapkan oleh permintaan lain!

依然出现了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、问题如图所示

Atas ialah kandungan terperinci Bagaimana Springboot mengintegrasikan Redis untuk mencapai masalah terlalu banyak jualan. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!