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

public class KafkaRetryRecord {

    public static final String KEY_RETRY_TIMES = "retryTimes";

    private String key;
    private String value;

    private Integer retryTimes;
    private String topic;
    private Long nextTime;

    public KafkaRetryRecord(){
    }

    public String getKey() {
        return key;
    }
    public void setKey(String key) {
        this.key = key;
    }
    public String getValue() {
        return value;
    }
    public void setValue(String value) {
        this.value = value;
    }
    public Integer getRetryTimes() {
        return retryTimes;
    }
    public void setRetryTimes(Integer retryTimes) {
        this.retryTimes = retryTimes;
    }
    public String getTopic() {
        return topic;
    }
    public void setTopic(String topic) {
        this.topic = topic;
    }
    public Long getNextTime() {
        return nextTime;
    }
    public void setNextTime(Long nextTime) {
        this.nextTime = nextTime;
    }

    public ProducerRecord parse(){
        Integer partition = null;
        Long timestamp = System.currentTimeMillis();
        List<Header> headers = new ArrayList<>();
        ByteBuffer retryTimesBuffer = ByteBuffer.allocate(4);
        retryTimesBuffer.putInt(retryTimes);
        retryTimesBuffer.flip();
        headers.add(new RecordHeader(KafkaRetryRecord.KEY_RETRY_TIMES, retryTimesBuffer));

        ProducerRecord sendRecord = new ProducerRecord(
                topic, partition, timestamp, key, value, headers);
        return sendRecord;
    }
}

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
62
63
64
65
66
67

public class KafkaRetryService {

	private static final Logger log = LoggerFactory.getLogger(KafkaRetryService.class);

	



	private static final int[] RETRY_INTERVAL_SECONDS = {10, 30, 1*60, 2*60, 5*60, 10*60, 30*60, 1*60*60, 2*60*60};

	


	@Value("${spring.kafka.topics.retry}")
	private String retryTopic;
	@Autowired
	private KafkaTemplate<String, String> template;

	public void consumerLater(ConsumerRecord<String, String> record){
		
		int retryTimes = getRetryTimes(record);
		Date nextConsumerTime = getNextConsumerTime(retryTimes);
		if(nextConsumerTime == null) {
			return;
		}

		KafkaRetryRecord retryRecord = new KafkaRetryRecord();
		retryRecord.setNextTime(nextConsumerTime.getTime());
		retryRecord.setTopic(record.topic());
		retryRecord.setRetryTimes(retryTimes);
		retryRecord.setKey(record.key());
		retryRecord.setValue(record.value());

		String value = JSON.toJSONString(retryRecord);
		template.send(retryTopic, null, value);
	}

	


	private int getRetryTimes(ConsumerRecord record){
		int retryTimes = -1;
		for(Header header : record.headers()){
			if(KafkaRetryRecord.KEY_RETRY_TIMES.equals(header.key())){
				ByteBuffer buffer = ByteBuffer.wrap(header.value());
				retryTimes = buffer.getInt();
			}
		}
		retryTimes++;
		return retryTimes;
	}

	


	private Date getNextConsumerTime(int retryTimes){
		
		if(RETRY_INTERVAL_SECONDS.length < retryTimes) {
			return null;
		}

		Calendar calendar = Calendar.getInstance();
		calendar.add(Calendar.SECOND, RETRY_INTERVAL_SECONDS[retryTimes]);
		return calendar.getTime();
	}

}

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

public class RetryListener {
	private static final Logger log = LoggerFactory.getLogger(RetryListener.class);

	private static final String RETRY_KEY_ZSET = "_retry_key";
	private static final String RETRY_VALUE_MAP = "_retry_value";
	@Autowired
	private RedisTemplate<String,Object> redisTemplate;
	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	@KafkaListener(topics = "${spring.kafka.topics.retry}")
	public void consume(List<ConsumerRecord<String, String>> list) {
		for(ConsumerRecord<String, String> record : list){
			KafkaRetryRecord retryRecord = JSON.parseObject(record.value(), KafkaRetryRecord.class);

			




			
			String key = UUID.randomUUID().toString();
			redisTemplate.opsForHash().put(RETRY_VALUE_MAP, key, record.value());
			redisTemplate.opsForZSet().add(RETRY_KEY_ZSET, key, retryRecord.getNextTime());
		}
	}

	


	@Scheduled(cron="0/2 * * * * *")
	public void retryFormRedis() {
		long currentTime = System.currentTimeMillis();
		Set<ZSetOperations.TypedTuple<Object>> typedTuples =
				redisTemplate.opsForZSet().reverseRangeByScoreWithScores(RETRY_KEY_ZSET, 0, currentTime);
		redisTemplate.opsForZSet().removeRangeByScore(RETRY_KEY_ZSET, 0, currentTime);
		for(ZSetOperations.TypedTuple<Object> tuple : typedTuples){
			String key = tuple.getValue().toString();
			String value = redisTemplate.opsForHash().get(RETRY_VALUE_MAP, key).toString();
			redisTemplate.opsForHash().delete(RETRY_VALUE_MAP, key);
			KafkaRetryRecord retryRecord = JSON.parseObject(value, KafkaRetryRecord.class);
			ProducerRecord record = retryRecord.parse();
			kafkaTemplate.send(record);
		}
		
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

public class ConsumeListener {
	private static final Logger log = LoggerFactory.getLogger(ConsumeListener.class);

	@Autowired
	private KafkaRetryService kafkaRetryService;

	@KafkaListener(topics = "${spring.kafka.topics.test}")
	public void consume(List<ConsumerRecord<String, String>> list) {
		for(ConsumerRecord record : list){
			try {
				
			} catch (Exception e){
				log.error(e.getMessage());
				
				kafkaRetryService.consumerLater(record);
			}
		}
	}

}