1
@Override
2
    public Result sendCode(String phone, HttpSession session) {
3
        // 1.電話番号の検証
4
        if (RegexUtils.isPhoneInvalid(phone)) {
5
            // 2.条件を満たさない場合、エラーメッセージを返す
6
            return Result.fail("手机号格式错误！");
7
        }
8
        // 3.条件を満たす場合、検証コードを生成
9
        String code = RandomUtil.randomNumbers(6);
10

11
        // 4.セッションに検証コードを保存
12
        session.setAttribute("code",code);
13
        // 5.検証コードを送信
14
        log.debug("发送短信验证码成功，验证码：{}", code);
15
        // OKを返す
16
        return Result.ok();
17
    }

1
@Override
2
    public Result login(LoginFormDTO loginForm, HttpSession session) {
3
        // 1.電話番号の検証
4
        String phone = loginForm.getPhone();
5
        if (RegexUtils.isPhoneInvalid(phone)) {
6
            // 2.条件を満たさない場合、エラーメッセージを返す
7
            return Result.fail("手机号格式错误！");
8
        }
9
        // 3.検証コードの検証
10
        Object cacheCode = session.getAttribute("code");
11
        String code = loginForm.getCode();
12
        if(cacheCode == null || !cacheCode.toString().equals(code)){
13
             //3.不一致、エラー
14
            return Result.fail("验证码错误");
15
        }
16
        // 一致、電話番号でユーザーを検索
17
        User user = query().eq("phone", phone).one();
18

19
        //5.ユーザーの存在を判断
20
        if(user == null){
21
            // 存在しない場合、作成
22
            user =  createUserWithPhone(phone);
23
        }
24
        //7.セッションにユーザー情報を保存
25
        session.setAttribute("user",user);
26

27
        return Result.ok();
28
    }

1
public class LoginInterceptor implements HandlerInterceptor {
2

3
    @Override
4
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
5
       //1.セッションを取得
6
        HttpSession session = request.getSession();
7
        //2.セッション内のユーザーを取得
8
        Object user = session.getAttribute("user");
9
        //3.ユーザーの存在を判定
10
        if(user == null){
11
              //4.存在しない場合、インターセプト
12
              response.setStatus(401);
13
              return false;
14
        }
15
        //5.存在する場合、ThreadLocalへ保存
16
        UserHolder.saveUser((User)user);
17
        //6.通過
18
        return true;
19
    }
20
}

1
@Configuration
2
public class MvcConfig implements WebMvcConfigurer {
3

4
    @Resource
5
    private StringRedisTemplate stringRedisTemplate;
6

7
    @Override
8
    public void addInterceptors(InterceptorRegistry registry) {
9
        // ログインインターセプター
10
        registry.addInterceptor(new LoginInterceptor())
11
                .excludePathPatterns(
12
                        "/shop/**",
13
                        "/voucher/**",
14
                        "/shop-type/**",
15
                        "/upload/**",
16
                        "/blog/hot",
17
                        "/user/code",
18
                        "/user/login"
19
                ).order(1);
20
        // トークン更新インターセプター
21
        registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).addPathPatterns("/**").order(0);
22
    }
23
}

1
//7.保存ユーザー情報をsessionへ
2
session.setAttribute("user", BeanUtils.copyProperties(user,UserDTO.class));
3

4
//5.存在時、ThreadLocalへユーザー情報を保存
5
UserHolder.saveUser((UserDTO) user);

1
@Override
2
public Result login(LoginFormDTO loginForm, HttpSession session) {
3
    // 1.電話番号の検証
4
    String phone = loginForm.getPhone();
5
    if (RegexUtils.isPhoneInvalid(phone)) {
6
        // 2.不適合の場合エラーを返す
7
        return Result.fail("手机号格式错误！");
8
    }
9
    // 3.Redisから検証コードを取得して検証
10
    String cacheCode = stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY + phone);
11
    String code = loginForm.getCode();
12
    if (cacheCode == null || !cacheCode.equals(code)) {
13
        // 不一致、エラー
14
        return Result.fail("验证码错误");
15
    }
16

17
    // 4.一致、電話番号でユーザーを検索
18
    User user = query().eq("phone", phone).one();
19

20
    // 5.判断して存在しなければ新規作成
21
    if (user == null) {
22
        user = createUserWithPhone(phone);
23
    }
24

25
    // 7.ユーザー情報をRedisに保存
26
    // 7.1.トークンをランダム生成、ログイントークンとして使用
27
    String token = UUID.randomUUID().toString(true);
28
    // 7.2.UserオブジェクトをHashMapへ変換して保存
29
    UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
30
    Map<String, Object> userMap = BeanUtil.beanToMap(userDTO, new HashMap<>(),
31
            CopyOptions.create()
32
                    .setIgnoreNullValue(true)
33
                    .setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString()));
34
    // 7.3.保存
35
    String tokenKey = LOGIN_USER_KEY + token;
36
    stringRedisTemplate.opsForHash().putAll(tokenKey, userMap);
37
    // 7.4.有効期限を設定
38
    stringRedisTemplate.expire(tokenKey, LOGIN_USER_TTL, TimeUnit.MINUTES);
39

40
    // 8.トークンを返す
41
    return Result.ok(token);
42
}

1
public class RefreshTokenInterceptor implements HandlerInterceptor {
2

3
    private StringRedisTemplate stringRedisTemplate;
4

5
    public RefreshTokenInterceptor(StringRedisTemplate stringRedisTemplate) {
6
        this.stringRedisTemplate = stringRedisTemplate;
7
    }
8

9
    @Override
10
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
11
        // 1.リクエストヘッダからtokenを取得
12
        String token = request.getHeader("authorization");
13
        if (StrUtil.isBlank(token)) {
14
            return true;
15
        }
16
        // 2.TOKENを基にRedisのユーザーを取得
17
        String key  = LOGIN_USER_KEY + token;
18
        Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(key);
19
        // 3.ユーザーが存在するか
20
        if (userMap.isEmpty()) {
21
            return true;
22
        }
23
        // 5.HashデータをUserDTOへ変換
24
        UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
25
        // 6.存在する場合、ThreadLocalへ保存
26
        UserHolder.saveUser(userDTO);
27
        // 7.トークン有効期限を更新
28
        stringRedisTemplate.expire(key, LOGIN_USER_TTL, TimeUnit.MINUTES);
29
        // 8.直ちにリクエストを通過
30
        return true;
31
    }
32

33
    @Override
34
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
35
        // ユーザーを削除
36
        UserHolder.removeUser();
37
    }
38
}

1
public class LoginInterceptor implements HandlerInterceptor {
2

3
    @Override
4
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
5
        // 1. ThreadLocalにユーザーが存在するかを判断
6
        if (UserHolder.getUser() == null) {
7
            // ない場合、401を設定して遮断
8
            response.setStatus(401);
9
            return false;
10
        }
11
        // ある場合、通過
12
        return true;
13
    }
14
}

1
@Override
2
public Result queryById(Long id) {
3
    String key = RedisConstants.CACHE_SHOP_KEY + id;
4
    String shopJson = stringRedisTemplate.opsForValue().get(key);
5

6
    if(StrUtil.isNotBlank(shopJson)) {
7
        Shop shop = JSONUtil.toBean(shopJson,Shop.class);
8
        return Result.ok(shop);
9
    }
10
    Shop shop = getById(id);
11
    if(shop == null) {
12
        return Result.fail("店铺不存在!");
13
    }
14
    stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),
15
            RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);
16

17
    return Result.ok(shop);
18
}

1
// クエリ時に期限を取得
2
stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),
3
        RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);
4

5
// 更新メソッドの追加
6
@Override
7
@Transactional
8
public Result update(Shop shop) {
9
    Long id = shop.getId();
10
    if (id == null) {
11
        return Result.fail("店铺id不能为空");
12
    }
13
    updateById(shop);
14

15
    stringRedisTemplate.delete(RedisConstants.CACHE_SHOP_KEY + id);
16
    return Result.ok();
17

18
}

1
@Override
2
public Result queryById(Long id) {
3
    String key = RedisConstants.CACHE_SHOP_KEY + id;
4
    String shopJson = stringRedisTemplate.opsForValue().get(key);
5

6
    if(StrUtil.isNotBlank(shopJson)) {
7
        Shop shop = JSONUtil.toBean(shopJson,Shop.class);
8
        return Result.ok(shop);
9
    }
10
    // キャッシュが空値かどうかを判定
11
    if(shopJson != null) {
12
        return Result.fail("店铺信息不存在");
13
    }
14

15
    Shop shop = getById(id);
16
    if(shop == null) {
17
        // 空値をキャッシュに書き込む
18
        stringRedisTemplate.opsForValue().set(key,"",
19
                RedisConstants.CACHE_NULL_TTL, TimeUnit.MINUTES);
20
        return Result.fail("店铺不存在!");
21
    }
22
    stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),
23
            RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);
24

25
    return Result.ok(shop);
26
}

1
@Slf4j
2
@Component
3
public class CacheClient {
4

5
    private final StringRedisTemplate stringRedisTemplate;
6

7
    private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);
8

9
    public CacheClient(StringRedisTemplate stringRedisTemplate) {
10
        this.stringRedisTemplate = stringRedisTemplate;
11
    }
12

13
    public void set(String key, Object value, Long time, TimeUnit unit) {
14
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value), time, unit);
15
    }
16

17
    public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit) {
18
        // 論理有効期限を設定
19
        RedisData redisData = new RedisData();
20
        redisData.setData(value);
21
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
22
        // Redisへ書き込み
23
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));
24
    }
25

26
    public <R,ID> R queryWithPassThrough(
27
            String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit){
28
        String key = keyPrefix + id;
29
        // 1.Redisからキャッシュを取得
30
        String json = stringRedisTemplate.opsForValue().get(key);
31
        // 2.存在判定
32
        if (StrUtil.isNotBlank(json)) {
33
            // 3.存在、型へ変換して返す
34
            return JSONUtil.toBean(json, type);
35
        }
36
        // キャッシュが空値かどうか
37
        if (json != null) {
38
            // エラーを返す
39
            return null;
40
        }
41

42
        // 4.存在しない場合、DBを照会
43
        R r = dbFallback.apply(id);
44
        // 5.存在しなければ、空値をキャッシュへ
45
        if (r == null) {
46
            stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
47
            return null;
48
        }
49
        // 6.存在する場合、Redisへ書き込み
50
        this.set(key, r, time, unit);
51
        return r;
52
    }
53

54
    public <R, ID> R queryWithLogicalExpire(
55
            String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {
56
        String key = keyPrefix + id;
57
        // 1.Redisからキャッシュを取得
58
        String json = stringRedisTemplate.opsForValue().get(key);
59
        // 2.存在判定
60
        if (StrUtil.isBlank(json)) {
61
            return null;
62
        }
63
        // 4.ヒット、先にJSONをオブジェクトへデシリアライズ
64
        RedisData redisData = JSONUtil.toBean(json, RedisData.class);
65
        R r = JSONUtil.toBean((JSONObject) redisData.getData(), type);
66
        LocalDateTime expireTime = redisData.getExpireTime();
67
        // 5.期限判定
68
        if(expireTime.isAfter(LocalDateTime.now())) {
69
            // 5.1 未期限、返す
70
            return r;
71
        }
72
        // 5.2 期限切れ、キャッシュ再構築
73
        // 6. キャッシュ再構築
74
        // 6.1 排他ロックを取得
75
        String lockKey = LOCK_SHOP_KEY + id;
76
        boolean isLock = tryLock(lockKey);
77
        // 6.2 ロックが取得できた場合、再構築を開始
78
        if (isLock){
79
            CACHE_REBUILD_EXECUTOR.submit(() -> {
80
                try {
81
                    // DB照会
82
                    R newR = dbFallback.apply(id);
83
                    // 再構築
84
                    this.setWithLogicalExpire(key, newR, time, unit);
85
                } catch (Exception e) {
86
                    throw new RuntimeException(e);
87
                }finally {
88
                    // ロックを解放
89
                    unlock(lockKey);
90
                }
91
            });
92
        }
93
        // 6.4 期限切れのデータを返す
94
        return r;
95
    }
96

97
    public <R, ID> R queryWithMutex(
98
            String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {
99
        String key = keyPrefix + id;
100
        // 1.Redisからキャッシュを取得
101
        String shopJson = stringRedisTemplate.opsForValue().get(key);
102
        // 2.存在判定
103
        if (StrUtil.isNotBlank(shopJson)) {
104
            // 3.存在、型へ変換して返す
105
            return JSONUtil.toBean(shopJson, type);
106
        }
107
        // キャッシュが空値かどうか
108
        if (shopJson != null) {
109
            // エラーを返す
110
            return null;
111
        }
112

113
        // 4.キャッシュ再構築
114
        // 4.1 排他ロックを取得
115
        String lockKey = LOCK_SHOP_KEY + id;
116
        R r = null;
117
        try {
118
            boolean isLock = tryLock(lockKey);
119
            // 4.2 取得成功か判定
120
            if (!isLock) {
121
                // 4.3 ロック取得失敗、待機して再試行
122
                Thread.sleep(50);
123
                return queryWithMutex(keyPrefix, id, type, dbFallback, time, unit);
124
            }
125
            // 4.4 ロック取得成功、DBを検索
126
            r = dbFallback.apply(id);
127
            // 5.存在しなければ、空値をRedisへ
128
            if (r == null) {
129
                stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
130
                return null;
131
            }
132
            // 6.存在する場合、Redisへ書き込み
133
            this.set(key, r, time, unit);
134
        } catch (InterruptedException e) {
135
            throw new RuntimeException(e);
136
        }finally {
137
            // 7.ロックを解放
138
            unlock(lockKey);
139
        }
140
        // 8.返す
141
        return r;
142
    }
143

144
    private boolean tryLock(String key) {
145
        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
146
        return BooleanUtil.isTrue(flag);
147
    }
148

149
    private void unlock(String key) {
150
        stringRedisTemplate.delete(key);
151
    }
152
}

1
@Component
2
public class RedisIdWorker {
3
    /**
4
     * 開始時間のタイムスタンプ
5
     */
6
    private static final long BEGIN_TIMESTAMP = 1640995200L;
7
    /**
8
     * シーケンスのビット数
9
     */
10
    private static final int COUNT_BITS = 32;
11

12
    private StringRedisTemplate stringRedisTemplate;
13

14
    public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {
15
        this.stringRedisTemplate = stringRedisTemplate;
16
    }
17

18
    public long nextId(String keyPrefix) {
19
        // 1. 時間スタンプを生成
20
        LocalDateTime now = LocalDateTime.now();
21
        long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
22
        long timestamp = nowSecond - BEGIN_TIMESTAMP;
23

24
        // 2. シーケンスを生成
25
        // 2.1 日付を日付で取得（日付まで）
26
        String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
27
        // 2.2 自動増分
28
        long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
29

30
        // 3. 結合して返却
31
        return timestamp << COUNT_BITS | count;
32
    }
33
}

1
@Override
2
@Transactional
3
public void addSeckillVoucher(Voucher voucher) {
4
    // クーポンを保存
5
    save(voucher);
6
    // 秒殺情報を保存
7
    SeckillVoucher seckillVoucher = new SeckillVoucher();
8
    seckillVoucher.setVoucherId(voucher.getId());
9
    seckillVoucher.setStock(voucher.getStock());
10
    seckillVoucher.setBeginTime(voucher.getBeginTime());
11
    seckillVoucher.setEndTime(voucher.getEndTime());
12
    seckillVoucherService.save(seckillVoucher);
13
    // 秒殺在庫をRedisに保存
14
    stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString());
15
}

1
@Override
2
public Result seckillVoucher(Long voucherId) {
3
    // 1.クーポンを検索
4
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
5
    // 2.秒殺が開始しているかを判定
6
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
7
        // 開始前
8
        return Result.fail("秒杀尚未开始！");
9
    }
10
    // 3.秒杀が終っているかを判定
11
    if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
12
        // 終了
13
        return Result.fail("秒杀已经结束！");
14
    }
15
    // 4.在庫が十分か
16
    if (voucher.getStock() < 1) {
17
        // 在庫不足
18
        return Result.fail("库存不足！");
19
    }
20
    //5.在庫をデクリメント
21
    boolean success = seckillVoucherService.update()
22
            .setSql("stock= stock -1")
23
            .eq("voucher_id", voucherId).update();
24
    if (!success) {
25
        // 在庫不足
26
        return Result.fail("库存不足！");
27
    }
28
    //6.注文を作成
29
    VoucherOrder voucherOrder = new VoucherOrder();
30
    // 6.1.注文ID
31
    long orderId = redisIdWorker.nextId("order");
32
    voucherOrder.setId(orderId);
33
    // 6.2.ユーザーID
34
    Long userId = UserHolder.getUser().getId();
35
    voucherOrder.setUserId(userId);
36
    // 6.3.クーポンID
37
    voucherOrder.setVoucherId(voucherId);
38
    save(voucherOrder);
39

40
    return Result.ok(orderId);
41
}

1
boolean success = seckillVoucherService.update()
2
    .setSql("stock= stock -1")
3
    .eq("voucher_id", voucherId)
4
    .gt("stock",0)
5
    .update(); //where id = ? and stock > 0

1
// 5. 一人一注文ロジック
2
// 5.1. ユーザーID
3
Long userId = UserHolder.getUser().getId();
4
int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
5
// 5.2. 存在判定
6
if (count > 0) {
7
    // ユーザーはすでに購入済み
8
    return Result.fail("用户已经购买过一次！");
9
}

1
// maven
2
<dependency>
3
    <groupId>org.aspectj</groupId>
4
    <artifactId>aspectjweaver</artifactId>
5
</dependency>
6

7
// Service
8
synchronized(userId.toString().intern()) {
9
    // 代理オブジェクトを取得（トランザクション）
10
    IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
11
    return proxy.createVoucherOrder(voucherId);
12
}
13

14
@Transactional
15
public Result createVoucherOrder(Long voucherId) {
16
    // 5.一人一注文ロジック
17
    // 5.1. ユーザーID
18
    Long userId = UserHolder.getUser().getId();
19

20
    int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
21
    // 5.2. 存在判定
22
    if (count > 0) {
23
        // ユーザーはすでに購入済み
24
        return Result.fail("用户已经购买过一次！");
25
    }
26

27
    //5、在庫を減らす
28
    boolean success = seckillVoucherService.update()
29
            .setSql("stock= stock -1")
30
            .eq("voucher_id", voucherId)
31
            .gt("stock",0)
32
            .update(); //where id = ? and stock > 0
33
    if (!success) {
34
        // 在庫不足
35
        return Result.fail("库存不足！");
36
    }
37
    //6. 注文を作成
38
    VoucherOrder voucherOrder = new VoucherOrder();
39
    // 6.1.注文ID
40
    long orderId = redisIdWorker.nextId("order");
41
    voucherOrder.setId(orderId);
42
    // 6.2.ユーザーID
43
    voucherOrder.setUserId(userId);
44
    // 6.3.クーポンID
45
    voucherOrder.setVoucherId(voucherId);
46
    save(voucherOrder);
47
    return Result.ok(orderId);
48

49
}

1
-- 1.パラメータリスト
2
-- 1.1. クーポンID
3
local voucherId = ARGV[1]
4
-- 1.2. ユーザーID
5
local userId = ARGV[2]
6
-- 1.3. 注文ID
7
local orderId = ARGV[3]
8

9
-- 2.データキー
10
-- 2.1. 在庫キー
11
local stockKey = 'seckill:stock:' .. voucherId
12
-- 2.2. 注文キー
13
local orderKey = 'seckill:order:' .. voucherId
14

15
-- 3.スクリプトのビジネス
16
-- 3.1. 在庫が充足しているか
17
if(tonumber(redis.call('get', stockKey)) <= 0) then
18
    -- 3.2. 在庫不足、1を返す
19
    return 1
20
end
21
-- 3.2. ユーザーが既に注文しているか
22
if(redis.call('sismember', orderKey, userId) == 1) then
23
    -- 3.3. 存在する、重複注文とみなし、2を返す
24
    return 2
25
end
26
-- 3.4. 在庫を減らす
27
redis.call('incrby', stockKey, -1)
28
-- 3.5. 注文（ユーザーを保存）sadd orderKey userId
29
redis.call('sadd', orderKey, userId)
30
-- 3.6. メッセージをキューへ送信、XADD stream.orders * k1 v1 k2 v2 ...
31
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
32
return 0

1
// 非同期処理用スレッドプール
2
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
3

4
// クラス初期化後に実行、クラスが初期化されたら、いつでも実行され得る
5
@PostConstruct
6
private void init() {
7
   SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
8
}
9
// スレッドプールで実行するタスク
10
// 初期化完了後、キューから情報を取得する
11
 private class VoucherOrderHandler implements Runnable{
12

13
        @Override
14
        public void run() {
15
            while (true){
16
                try {
17
                    // 1.キューの注文情報を取得 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
18
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
19
                    Consumer.from("g1", "c1"),
20
                    StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
21
                    StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
22
                );
23
                // 2.注文情報が空かを判定
24
                if (list == null || list.isEmpty()) {
25
                    // nullの場合、メッセージが無いとみなし、次のループへ
26
                    continue;
27
                }
28
                // データを解析
29
                MapRecord<String, Object, Object> record = list.get(0);
30
                Map<Object, Object> value = record.getValue();
31
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
32
                // 3.注文を作成
33
                createVoucherOrder(voucherOrder);
34
                // 4.メッセージの確認 XACK
35
                stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
36
            } catch (Exception e) {
37
                log.error("处理订单异常", e);
38
                // 異常メッセージを処理
39
                handlePendingList();
40
            }
41
        }
42
    }
43

44
    private void handlePendingList() {
45
        while (true) {
46
            try {
47
                // 1. pending-list の情報を取得 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0
48
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
49
                    Consumer.from("g1", "c1"),
50
                    StreamReadOptions.empty().count(1),
51
                    StreamOffset.create("stream.orders", ReadOffset.from("0"))
52
                );
53
                // 2.注文情報が空かを判定
54
                if (list == null || list.isEmpty()) {
55
                    // nullの場合、異常データが無いので終了
56
                    break;
57
                }
58
                // データを解析
59
                MapRecord<String, Object, Object> record = list.get(0);
60
                Map<Object, Object> value = record.getValue();
61
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
62
                // 3.注文を作成
63
                createVoucherOrder(voucherOrder);
64
                // 4.メッセージの確認 XACK
65
                stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
66
            } catch (Exception e) {
67
                log.error("处理pendding订单异常", e);
68
                try{
69
                    Thread.sleep(20);
70
                }catch(Exception e){
71
                    e.printStackTrace();
72
                }
73
            }
74
        }
75
    }
76
}