java

说明

通过rightPopAndLeftPush原子操作在拿出消息前将消息先推到另一个补偿队列中，保证消息可靠性
通过xxljob等定时任务轮询实现补偿
消息设置通知时间字段以实现补偿5分钟后的通知
为什么补偿不使用spring的定时任务？因为分布式轮询会导致重复处理，xxljob设置单机执行则不会

redis配置

1
2
3
4
5
6
7
8
9
10
11
12
spring:
	redis:
	  host: 127.0.0.1
	  port: 3379
	  password: "abc123"
	  database: 1
	  pool:
	    max-active: 1000
	    max-wait: 60000
	    min-idle: 20
	    max-idle: 20
	  timeout: 2000

使用

仅添加补偿机制

实体类

1
2
3
4
public class NotifyDto {  
    private String orderNo;  
    private long notifyTime;  
}

生产端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import org.springframework.data.redis.core.StringRedisTemplate;
import org.apache.commons.lang3.StringUtils;
import com.alibaba.fastjson.JSON;

@Resource  
private StringRedisTemplate stringRedisTemplate;

public boolean push(String notify) {  
    if (StringUtils.isEmpty(notify)) {  
        throw new BusinessException("通知内容为空，接收失败");  
    }  
  
    log.info("收到通知：{}", notify);
    NotifyDto notifyDto = JSON.parseObject(notify, NotifyDto.class);
	notifyDto.setNotifyTime(Instant.now().toEpochMilli());
  
    // 同步保存到redis  
    Long l = stringRedisTemplate.opsForList().leftPush(QUEUE_NAME, JSON.toJSONString(notifyDto));  
    return l != null;  
}

消费端

从队列中取出后先放入pending队列中，等处理完之后再删除，如果系统宕机可以接着处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import org.springframework.data.redis.core.StringRedisTemplate;
import com.alibaba.fastjson.JSON;

@Resource  
private StringRedisTemplate stringRedisTemplate;

@Scheduled(fixedDelay = 100)  
public void pop() {  

	Long size = stringRedisTemplate.opsForList().size(QUEUE_NAME);  
	if (size == null || size == 0) {  
	    return;  
	}

    String notify = stringRedisTemplate.opsForList()  
            .rightPopAndLeftPush(QUEUE_NAME, QUEUE_NAME_PENDING);  
  
    if(notify == null) {
        return;  
    }  
    
    handleNotify(notifyDto, notify);  
}

/**
 * 补偿消费
 * 每五分钟扫描一次
 * 消息已经存在队列超过30分钟，则不再处理
 * 如果消息已经存在队列超过5分钟，则进行处理
 **/
@XxlJob("pendingNotify")  
public void pendingNotify() {  
  
    List<String> notifyList = stringRedisTemplate.opsForList().range(QUEUE_NAME_PENDING, 0, -1);  
    if (notifyList == null || notifyList.isEmpty()) {  
        return;  
    }  
    for (String notify : notifyList) {   
        log.info("开始处理通知：{}", notify);  
        NotifyDto notifyDto = JSON.parseObject(notify, NotifyDto.class);  
        Instant now = Instant.now();  
        Instant notifyTime = Instant.ofEpochMilli(notifyDto.getNotifyTime());  
        Instant after5m = notifyTime.plus(5, ChronoUnit.MINUTES);  
        Instant after30m = notifyTime.plus(30, ChronoUnit.MINUTES);  
        // 超过30分钟不再处理  
        if(after30m.isAfter(now)) {  
            stringRedisTemplate.opsForList().remove(QUEUE_NAME_PENDING, 0, notify);  
            return;  
        }  
        // 只处理超过5分钟的消息
        if(after5m.isBefore(now)) {  
            handleNotify(notifyDto, notify);  
        }  
    }  
}

private void handleNotify(NotifyDto notifyDto, String notify){
	// do something...
	// 处理完成移除消息
	stringRedisTemplate.opsForList().remove(QUEUE_NAME_PENDING, 0, notify);
}

补偿优化

bean

1
2
3
4
5
6
7
8
9
```java
public class NotifyDto {  
    private String orderNo;  
    private long notifyTime;  
    /**  
     * 原始通知字符串  
     */  
    private String originNotify;  
}

job

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@XxlJob("pendingNotify")  
public ReturnT<String> pendingNotify(String args) {  
    Long size = stringRedisTemplate.opsForList().size(QUEUE_NAME_PENDING);  
    if (size == null || size == 0) {  
        return ReturnT.SUCCESS;  
    }  
  
    List<String> notifyList = stringRedisTemplate.opsForList()  
            .range(TicketOrderConstant.QUEUE_NAME_PENDING, 0, -1);  
    if (notifyList == null || notifyList.isEmpty()) {  
        return ReturnT.SUCCESS;  
    }  
  
    // 解析并过滤  
    List<NotifyDto> filteredList = notifyList.stream()  
            .map(notify -> {  
                NotifyDto notifyDto = JSON.parseObject(notify, NotifyDto.class);  
                notifyDto.setOriginNotify(notify);  
                return notifyDto;  
            })  
            .filter(notifyDto -> {  
                Instant now = Instant.now();  
                Instant notifyTime = Instant.ofEpochMilli(notifyDto.getNotifyTime());  
                Instant after5m = notifyTime.plus(5, ChronoUnit.MINUTES);  
                Instant after1h = notifyTime.plus(1, ChronoUnit.HOURS);  
                // 超过60分钟的不再处理  
                if (now.isAfter(after1h)) {  
                    return false;  
                }  
                // 只处理超过5分钟的通知  
                return now.isAfter(after5m);  
            }).collect(Collectors.toList());  
  
    // 补偿
    filteredList.forEach(notifyDto -> {  
        try {  
            handleNotify(notifyDto);  
        } catch (Exception e) {  
            log.error("补偿失败");  
        }  
    });  
    return ReturnT.SUCCESS;  
}

private void handleNotify(NotifyDto notifyDto){
	// do something...
	// 处理完成移除消息
	stringRedisTemplate.opsForList().remove(QUEUE_NAME_PENDING, 0, notifyDto.getOriginNotify());
}

时间节点补偿方式

消息设置时间节点字段以实现每补偿失败一次，延顺到下个时间节点

bean

1
2
3
4
5
6
7
8
9
10
11
12
public class NotifyDto {  
    private long notifyTime;  
    private String orderNo;  
    /**  
     * 时间梯度，每失败一次依次递增  
     */  
    private int timeStep = 0;  
    /**  
     * 原始通知字符串  
     */  
    private String originNotify;  
}

job

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
/**
 * 单位：分
 */
public static final int[] TIME_STEP_ARR = new int[]{5, 15, 30, 60, 180, 360, 720, 1440};

@XxlJob("refundAmount")  
public ReturnT<String> refundAmount(String args) {  
    Long size = stringRedisTemplate.opsForList().size(QUEUE_NAME_PENDING);  
    if (size == null || size == 0) {  
        return ReturnT.SUCCESS;  
    }  
  
    List<String> notifyList = stringRedisTemplate.opsForList().range(QUEUE_NAME_PENDING, 0, -1);  
    if (notifyList == null || notifyList.isEmpty()) {  
        return ReturnT.SUCCESS;  
    }  
  
    // 转换消息并且过滤出指定时间范围内的消息  
    List<NotifyDto> refundList = notifyList.stream()  
            .map(s -> {  
                NotifyDto notifyDto = JSON.parseObject(s, NotifyDto.class);  
                notifyDto.setOriginNotify(s);  
                return notifyDto;  
            })  
            .filter(notifyDto -> {  
                Instant now = Instant.now();  
                Instant notifyTime = Instant.ofEpochMilli(notifyDto.getNotifyTime());  
                int stepGap = TIME_STEP_ARR[notifyDto.getTimeStep()];  
                Instant stepTime = notifyTime.plus(stepGap, ChronoUnit.MINUTES);  
                return notifyDto.getTimeStep() < 8 && now.isAfter(stepTime);  
            }).collect(Collectors.toList());  
  
    // 处理消息  
    for (NotifyDto notifyDto : refundList) {  
        try {  
            log.info("开始处理未处理的消息：{}", notifyDto.getOriginNotify());  
            handleNotify(notifyDto);  
        } catch (Exception e) {  
            log.error("消息处理失败：{}", notifyDto.getOriginNotify(), e);  
            timeStepForward(notifyDto);  
        }  
    }  
    return ReturnT.SUCCESS;  
}

private void timeStepForward(NotifyDto notifyDto) {  
    notifyDto.setTimeStep(notifyDto.getTimeStep() + 1);  
    String originNotify = notifyDto.getOriginNotify();  
    notifyDto.setOriginNotify(null);  
    // 保证原子性  
    try {  
        stringRedisTemplate.setEnableTransactionSupport(true);  
        stringRedisTemplate.multi();  
        stringRedisTemplate.opsForList().leftPush(QUEUE_NAME_PENDING, JSON.toJSONString(notifyDto));  
        stringRedisTemplate.opsForList().remove(QUEUE_NAME_PENDING, 0, originNotify);  
        stringRedisTemplate.exec();  
    } finally {  
        stringRedisTemplate.setEnableTransactionSupport(false);  
    }  
}

Leave a comment

waline