devbox
秋刀鱼在做梦
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

4.2.8 Kafka 延时队列, 重试队列(结合redis实现)_fetch.min.isr

作者：秋刀鱼在做梦 | 2024-07-08 13:34:33

赞

踩

fetch.min.isr

目录

2.7 延时队列

2.8 重试队列

代码实现

 

Kafka 高级特性-延时/重试队列

 

2.7 延时队列

两个follower副本都已经拉取到了leader副本的最新位置，此时又向leader副本发送拉取请求，而leader副本并没有新的消息写入，那么此时leader副本该如何处理呢？可以直接返回空的拉取结果给follower副本，不过在leader副本一直没有新消息写入的情况下，follower副本会一直发送拉取请求，并且总收到空的拉取结果，消耗资源。

Kafka在处理拉取请求时，会先读取一次日志文件，如果收集不到足够多（fetchMinBytes，由参数fetch.min.bytes配置，默认值为1）的消息，那么就会创建一个延时拉取操作（DelayedFetch）以等待拉取到足够数量的消息。当延时拉取操作执行时，会再读取一次日志文件，然后将拉取结果返回给follower副本。

延迟操作不只是拉取消息时的特有操作，在Kafka中有多种延时操作，比如延时数据删除、延时生产等。

对于延时生产（消息）而言，如果在使用生产者客户端发送消息的时候将acks参数设置为-1，那么就意味着需要等待ISR集合中的所有副本都确认收到消息之后才能正确地收到响应的结果，或者捕获超时异常。

 

 

假设某个分区有3个副本：leader、follower1和follower2，它们都在分区的ISR集合中。不考虑ISR变动的情况，Kafka在收到客户端的生产请求后，将消息3和消息4写入leader副本的本地日志文件。

由于客户端设置了acks为-1，那么需要等到follower1和follower2两个副本都收到消息3和消息4后才能告知客户端正确地接收了所发送的消息。如果在一定的时间内，follower1副本或follower2副本没能够完全拉取到消息3和消息4，那么就需要返回超时异常给客户端。生产请求的超时时间由参数request.timeout.ms配置，默认值为30000，即30s。

那么这里等待消息3和消息4写入follower1副本和follower2副本，并返回相应的响应结果给客户端的动作是由谁来执行的呢？在将消息写入leader副本的本地日志文件之后，Kafka会创建一个延时的生产操作（DelayedProduce），用来处理消息正常写入所有副本或超时的情况，以返回相应的响应结果给客户端。

延时操作需要延时返回响应的结果，首先它必须有一个超时时间（delayMs），如果在这个超时时间内没有完成既定的任务，那么就需要强制完成以返回响应结果给客户端。其次，延时操作不同于定时操作，定时操作是指在特定时间之后执行的操作，而延时操作可以在所设定的超时时间之前完成，所以延时操作能够支持外部事件的触发。

就延时生产操作而言，它的外部事件是所要写入消息的某个分区的HW（高水位）发生增长。也就是说，随着follower副本不断地与leader副本进行消息同步，进而促使HW进一步增长，HW每增长一次都会检测是否能够完成此次延时生产操作，如果可以就执行以此返回响应结果给客户端；如果在超时时间内始终无法完成，则强制执行。

延时拉取操作，是由超时触发或外部事件触发而被执行的。超时触发很好理解，就是等到超时时间之后触发第二次读取日志文件的操作。外部事件触发就稍复杂了一些，因为拉取请求不单单由follower副本发起，也可以由消费者客户端发起，两种情况所对应的外部事件也是不同的。如果是follower副本的延时拉取，它的外部事件就是消息追加到了leader副本的本地日志文件中；如果是消费者客户端的延时拉取，它的外部事件可以简单地理解为HW的增长。

 

时间轮实现延时队列。
TimeWheel。size，每个单元格的时间，   每个单元格都代表一个时间，size*每个单元格的时间就是一个周期。

 

 

 

2.8 重试队列

kafka没有重试机制不支持消息重试，也没有死信队列，因此使用kafka做消息队列时，需要自己实现消息重试的功能。

实现

创建新的kafka主题作为重试队列：
1. 创建一个topic作为重试topic，用于接收等待重试的消息。
2. 普通topic消费者设置待重试消息的下一个重试topic。
3. 从重试topic获取待重试消息储存到redis的zset中，并以下一次消费时间排序
4. 定时任务从redis获取到达消费事件的消息，并把消息发送到对应的topic
5. 同一个消息重试次数过多则不再重试

 

代码实现

1. 新建springboot项目

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3. 		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4. 	<modelVersion>4.0.0</modelVersion>
  5. 	<parent>
  6. 		<groupId>org.springframework.boot</groupId>
  7. 		<artifactId>spring-boot-starter-parent</artifactId>
  8. 		<version>2.2.8.RELEASE</version>
  9. 		<relativePath/> <!-- lookup parent from repository -->
  10. 	</parent>
  11. 	<groupId>com.lagou.kafka.demo</groupId>
  12. 	<artifactId>demo-retryqueue</artifactId>
  13. 	<version>0.0.1-SNAPSHOT</version>
  14. 	<name>demo-retryqueue</name>
  15. 	<description>Demo project for Spring Boot</description>
  16.  
  17. 	<properties>
  18. 		<java.version>1.8</java.version>
  19. 	</properties>
  20.  
  21. 	<dependencies>
  22. 		<dependency>
  23. 			<groupId>org.springframework.boot</groupId>
  24. 			<artifactId>spring-boot-starter-data-redis</artifactId>
  25. 		</dependency>
  26. 		<dependency>
  27. 			<groupId>org.springframework.boot</groupId>
  28. 			<artifactId>spring-boot-starter-web</artifactId>
  29. 		</dependency>
  30. 		<dependency>
  31. 			<groupId>org.springframework.kafka</groupId>
  32. 			<artifactId>spring-kafka</artifactId>
  33. 		</dependency>
  34. 		<dependency>
  35. 			<groupId>com.alibaba</groupId>
  36. 			<artifactId>fastjson</artifactId>
  37. 			<version>1.2.73</version>
  38. 		</dependency>
  39. 		<dependency>
  40. 			<groupId>org.springframework.boot</groupId>
  41. 			<artifactId>spring-boot-starter-test</artifactId>
  42. 			<scope>test</scope>
  43. 			<exclusions>
  44. 				<exclusion>
  45. 					<groupId>org.junit.vintage</groupId>
  46. 					<artifactId>junit-vintage-engine</artifactId>
  47. 				</exclusion>
  48. 			</exclusions>
  49. 		</dependency>
  50. 		<dependency>
  51. 			<groupId>io.projectreactor</groupId>
  52. 			<artifactId>reactor-test</artifactId>
  53. 			<scope>test</scope>
  54. 		</dependency>
  55. 		<dependency>
  56. 			<groupId>org.springframework.kafka</groupId>
  57. 			<artifactId>spring-kafka-test</artifactId>
  58. 			<scope>test</scope>
  59. 		</dependency>
  60. 	</dependencies>
  61.  
  62. 	<build>
  63. 		<plugins>
  64. 			<plugin>
  65. 				<groupId>org.springframework.boot</groupId>
  66. 				<artifactId>spring-boot-maven-plugin</artifactId>
  67. 			</plugin>
  68. 		</plugins>
  69. 	</build>
  70.  
  71. </project>

2. 添加application.properties

  1. # bootstrap.servers
  2. spring.kafka.bootstrap-servers=node1:9092
  3. # key序列化器
  4. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  5. # value序列化器
  6. spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
  7.  
  8. # 消费组id：group.id
  9. spring.kafka.consumer.group-id=retryGroup
  10. # key反序列化器
  11. spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  12. # value反序列化器
  13. spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  14.  
  15. # redis数据库编号
  16. spring.redis.database=0
  17. # redis主机地址
  18. spring.redis.host=node1
  19. # redis端口
  20. spring.redis.port=6379
  21. # Redis服务器连接密码（默认为空）
  22. spring.redis.password=
  23. # 连接池最大连接数（使用负值表示没有限制）
  24. spring.redis.jedis.pool.max-active=20
  25. # 连接池最大阻塞等待时间（使用负值表示没有限制）
  26. spring.redis.jedis.pool.max-wait=-1
  27. # 连接池中的最大空闲连接
  28. spring.redis.jedis.pool.max-idle=10
  29. # 连接池中的最小空闲连接
  30. spring.redis.jedis.pool.min-idle=0
  31. # 连接超时时间（毫秒）
  32. spring.redis.timeout=1000
  33.  
  34. # Kafka主题名称，业务主题
  35. spring.kafka.topics.test=tp_demo_retry_01
  36. # 重试队列，重试主题
  37. spring.kafka.topics.retry=tp_demo_retry_02

3. RetryqueueApplication.java

  1. package com.lagou.kafka.demo;
  2.  
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5.  
  6. @SpringBootApplication
  7. public class RetryqueueApplication {
  8.  
  9.     public static void main(String[] args) {
  10.         SpringApplication.run(RetryqueueApplication.class, args);
  11.     }
  12.  
  13. }

4. AppConfig.java

  1. package com.lagou.kafka.demo.config;
  2.  
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.data.redis.connection.RedisConnectionFactory;
  6. import org.springframework.data.redis.core.RedisTemplate;
  7.  
  8. // 配置redis template
  9. @Configuration
  10. public class AppConfig {
  11.     @Bean
  12.     public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
  13.  
  14.         RedisTemplate<String, Object> template = new RedisTemplate<>();
  15.         // 配置连接工厂
  16.         template.setConnectionFactory(factory);
  17.  
  18.         return template;
  19.     }
  20.  
  21. }

5. KafkaController.java

  1. package com.lagou.kafka.demo.controller;
  2.  
  3. import com.lagou.kafka.demo.service.KafkaService;
  4. import org.apache.kafka.clients.producer.ProducerRecord;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.beans.factory.annotation.Value;
  7. import org.springframework.web.bind.annotation.PathVariable;
  8. import org.springframework.web.bind.annotation.RequestMapping;
  9. import org.springframework.web.bind.annotation.RestController;
  10.  
  11. import java.util.concurrent.ExecutionException;
  12.  
  13. @RestController
  14. public class RetryController {
  15.  
  16.     @Autowired
  17.     private KafkaService kafkaService;
  18.  
  19.     @Value("${spring.kafka.topics.test}")
  20.     private String topic;
  21.  
  22.     @RequestMapping("/send/{message}")
  23.     public String sendMessage(@PathVariable String message) throws ExecutionException, InterruptedException {
  24.  
  25.         ProducerRecord<String, String> record = new ProducerRecord<>(
  26.                 topic,
  27.                 message
  28.         );
  29.  
  30.         // 向业务主题发送消息
  31.         String result = kafkaService.sendMessage(record);
  32.  
  33.         return result;
  34.     }
  35.  
  36. }

6. KafkaService.java

  1. package com.lagou.kafka.demo.service;
  2.  
  3. import org.apache.kafka.clients.producer.ProducerRecord;
  4. import org.apache.kafka.clients.producer.RecordMetadata;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.kafka.core.KafkaTemplate;
  9. import org.springframework.kafka.support.SendResult;
  10. import org.springframework.stereotype.Service;
  11.  
  12. import java.util.concurrent.ExecutionException;
  13.  
  14. @Service
  15. public class KafkaService {
  16.  
  17.     private Logger log = LoggerFactory.getLogger(KafkaService.class);
  18.  
  19.     // 标红不管
  20.     @Autowired
  21.     private KafkaTemplate<String, String> kafkaTemplate;
  22.  
  23.     public String sendMessage(ProducerRecord<String, String> record) throws ExecutionException, InterruptedException {
  24.  
  25.         SendResult<String, String> result = this.kafkaTemplate.send(record).get();
  26.         RecordMetadata metadata = result.getRecordMetadata();
  27.         String returnResult = metadata.topic() + "\t" + metadata.partition() + "\t" + metadata.offset();
  28.         log.info("发送消息成功：" + returnResult);
  29.  
  30.         return returnResult;
  31.     }
  32.  
  33. }

7. ConsumerListener.java

  1. package com.lagou.kafka.demo.listener;
  2.  
  3. import com.lagou.kafka.demo.service.RetryService;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.kafka.annotation.KafkaListener;
  9. import org.springframework.stereotype.Component;
  10.  
  11. @Component
  12. public class ConsumerListener {
  13.  
  14.     private static final Logger log = LoggerFactory.getLogger(ConsumerListener.class);
  15.  
  16.     @Autowired
  17.     private RetryService kafkaRetryService;
  18.  
  19.     private static int index = 0;
  20.  
  21.  
  22.     // 拉取下面主题的消息
  23.     @KafkaListener(topics = "${spring.kafka.topics.test}", groupId = "${spring.kafka.consumer.group-id}")
  24.     public void consume(ConsumerRecord<String, String> record) {
  25.         try {
  26.             // 业务处理
  27.             log.info("消费的消息：" + record);
  28.             index++;
  29.             if (index % 2 == 0) {
  30.                 throw new Exception("该重发了");
  31.             }
  32.         } catch (Exception e) {
  33.             log.error(e.getMessage());
  34.             // 消息重试，实际上先将消息放到redis, 再从redis放到消息队列中
  35.             kafkaRetryService.consumerLater(record);
  36.         }
  37.     }
  38.  
  39. }

8. KafkaRetryService.java

  1. package com.lagou.kafka.demo.service;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.lagou.kafka.demo.entity.RetryRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecord;
  6. import org.apache.kafka.common.header.Header;
  7. import org.slf4j.Logger;
  8. import org.slf4j.LoggerFactory;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.beans.factory.annotation.Value;
  11. import org.springframework.kafka.core.KafkaTemplate;
  12. import org.springframework.stereotype.Service;
  13.  
  14. import java.nio.ByteBuffer;
  15. import java.util.Calendar;
  16. import java.util.Date;
  17.  
  18. @Service
  19. public class RetryService {
  20.     private static final Logger log = LoggerFactory.getLogger(RetryService.class);
  21.  
  22.     /**
  23.      * 消息消费失败后下一次消费的延迟时间(秒)
  24.      * 第一次重试延迟10秒;第	二次延迟30秒,第三次延迟1分钟...
  25.      */
  26.     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};
  27.  
  28.     /**
  29.      * 重试topic
  30.      */
  31.     @Value("${spring.kafka.topics.retry}")
  32.     private String retryTopic;
  33.  
  34.     @Autowired
  35.     private KafkaTemplate<String, String> kafkaTemplate;
  36.  
  37.     public void consumerLater(ConsumerRecord<String, String> record){
  38.         // 获取消息的已重试次数
  39.         int retryTimes = getRetryTimes(record);
  40.         Date nextConsumerTime = getNextConsumerTime(retryTimes);
  41.         // 如果达到重试次数，则不再重试
  42.         if(nextConsumerTime == null) {
  43.             return;
  44.         }
  45.  
  46.         // 组织消息
  47.         RetryRecord retryRecord = new RetryRecord();
  48.         retryRecord.setNextTime(nextConsumerTime.getTime());
  49.         retryRecord.setTopic(record.topic());
  50.         retryRecord.setRetryTimes(retryTimes);
  51.         retryRecord.setKey(record.key());
  52.         retryRecord.setValue(record.value());
  53.  
  54.         // 转换为字符串
  55.         String value = JSON.toJSONString(retryRecord);
  56.         // 发送到重试队列
  57.         kafkaTemplate.send(retryTopic, null, value);
  58.     }
  59.  
  60.     /**
  61.      * 获取消息的已重试次数
  62.      */
  63.     private int getRetryTimes(ConsumerRecord record){
  64.         int retryTimes = -1;
  65.         for(Header header : record.headers()){
  66.             if(RetryRecord.KEY_RETRY_TIMES.equals(header.key())){
  67.                 ByteBuffer buffer = ByteBuffer.wrap(header.value());
  68.                 retryTimes = buffer.getInt();
  69.             }
  70.         }
  71.         retryTimes++;
  72.         return retryTimes;
  73.     }
  74.  
  75.     /**
  76.      * 获取待重试消息的下一次消费时间
  77.      */
  78.     private Date getNextConsumerTime(int retryTimes){
  79.         // 重试次数超过上限,不再重试
  80.         if(RETRY_INTERVAL_SECONDS.length < retryTimes) {
  81.             return null;
  82.         }
  83.  
  84.         Calendar calendar = Calendar.getInstance();
  85.         calendar.add(Calendar.SECOND, RETRY_INTERVAL_SECONDS[retryTimes]);
  86.         return calendar.getTime();
  87.     }
  88. }

9. RetryListener.java

  1. package com.lagou.kafka.demo.listener;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.lagou.kafka.demo.entity.RetryRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecord;
  6. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import org.slf4j.Logger;
  8. import org.slf4j.LoggerFactory;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.beans.factory.annotation.Value;
  11. import org.springframework.data.redis.core.RedisTemplate;
  12. import org.springframework.data.redis.core.ZSetOperations;
  13. import org.springframework.kafka.annotation.KafkaListener;
  14. import org.springframework.kafka.core.KafkaTemplate;
  15. import org.springframework.scheduling.annotation.EnableScheduling;
  16. import org.springframework.scheduling.annotation.Scheduled;
  17. import org.springframework.stereotype.Component;
  18.  
  19. import java.util.Set;
  20. import java.util.UUID;
  21.  
  22. @Component
  23. // 下面注解表示开启调度
  24. @EnableScheduling
  25. public class RetryListener {
  26.  
  27.     private Logger log = LoggerFactory.getLogger(RetryListener.class);
  28.  
  29.     private static final String RETRY_KEY_ZSET = "_retry_key";      // 时间
  30.     private static final String RETRY_VALUE_MAP = "_retry_value";   // 消息
  31.     @Autowired
  32.     private RedisTemplate<String,Object> redisTemplate;
  33.     @Autowired
  34.     private KafkaTemplate<String, String> kafkaTemplate;
  35.  
  36.     //
  37.     @Value("${spring.kafka.topics.test}")
  38.     private String bizTopic;
  39.  
  40.     @KafkaListener(topics = "${spring.kafka.topics.retry}")     // 只要有消息就取出来
  41. //    public void consume(List<ConsumerRecord<String, String>> list) {
  42. //        for(ConsumerRecord<String, String> record : list){
  43.     public void consume(ConsumerRecord<String, String> record) {
  44.  
  45.         System.out.println("需要重试的消息：" + record);
  46.         RetryRecord retryRecord = JSON.parseObject(record.value(), RetryRecord.class);
  47.  
  48.         /**
  49.          * 防止待重试消息太多撑爆redis,可以将待重试消息按下一次重试时间分开存储放到不同介质
  50.          * 例如下一次重试时间在半小时以后的消息储存到mysql,并定时从mysql读取即将重试的消息储储存到redis
  51.          */
  52.  
  53.         // 通过redis的zset进行时间排序
  54.         String key = UUID.randomUUID().toString();
  55.         redisTemplate.opsForHash().put(RETRY_VALUE_MAP, key, record.value());
  56.         redisTemplate.opsForZSet().add(RETRY_KEY_ZSET, key, retryRecord.getNextTime());
  57.     }
  58. //    }
  59.  
  60.     /**
  61.      * 定时任务从redis读取到达重试时间的消息,发送到对应的topic
  62.      */
  63. //    @Scheduled(cron="2 * * * * *")
  64.     @Scheduled(fixedDelay = 2000)
  65.     public void retryFromRedis() {
  66.         log.warn("retryFromRedis----begin");
  67.         long currentTime = System.currentTimeMillis();
  68.         // 根据时间倒序获取
  69.         Set<ZSetOperations.TypedTuple<Object>> typedTuples =
  70.                 redisTemplate.opsForZSet().reverseRangeByScoreWithScores(RETRY_KEY_ZSET, 0, currentTime);
  71.         // 移除取出的消息
  72.         redisTemplate.opsForZSet().removeRangeByScore(RETRY_KEY_ZSET, 0, currentTime);
  73.         for(ZSetOperations.TypedTuple<Object> tuple : typedTuples){
  74.             String key = tuple.getValue().toString();
  75.             String value = redisTemplate.opsForHash().get(RETRY_VALUE_MAP, key).toString();
  76.             redisTemplate.opsForHash().delete(RETRY_VALUE_MAP, key);
  77.             RetryRecord retryRecord = JSON.parseObject(value, RetryRecord.class);
  78.             ProducerRecord record = retryRecord.parse();
  79.  
  80.             ProducerRecord recordReal = new ProducerRecord(
  81.                     bizTopic,
  82.                     record.partition(),
  83.                     record.timestamp(),
  84.                     record.key(),
  85.                     record.value(),
  86.                     record.headers()
  87.             );
  88.  
  89.             kafkaTemplate.send(recordReal);
  90.         }
  91.         // todo 发生异常将发送失败的消息重新发送到redis
  92.     }
  93. }

10. RetryRecord.java

  1. package com.lagou.kafka.demo.entity;
  2.  
  3. import org.apache.kafka.clients.producer.ProducerRecord;
  4. import org.apache.kafka.common.header.Header;
  5. import org.apache.kafka.common.header.internals.RecordHeader;
  6.  
  7. import java.nio.ByteBuffer;
  8. import java.util.ArrayList;
  9. import java.util.List;
  10.  
  11. public class RetryRecord {
  12.  
  13.     public static final String KEY_RETRY_TIMES = "retryTimes";
  14.  
  15.     private String key;
  16.     private String value;
  17.  
  18.     private Integer retryTimes;
  19.     private String topic;
  20.     private Long nextTime;
  21.  
  22.     public RetryRecord() {
  23.     }
  24.  
  25.     public String getKey() {
  26.         return key;
  27.     }
  28.  
  29.     public void setKey(String key) {
  30.         this.key = key;
  31.     }
  32.  
  33.     public String getValue() {
  34.         return value;
  35.     }
  36.  
  37.     public void setValue(String value) {
  38.         this.value = value;
  39.     }
  40.  
  41.     public Integer getRetryTimes() {
  42.         return retryTimes;
  43.     }
  44.  
  45.     public void setRetryTimes(Integer retryTimes) {
  46.         this.retryTimes = retryTimes;
  47.     }
  48.  
  49.     public String getTopic() {
  50.         return topic;
  51.     }
  52.  
  53.     public void setTopic(String topic) {
  54.         this.topic = topic;
  55.     }
  56.  
  57.     public Long getNextTime() {
  58.         return nextTime;
  59.     }
  60.  
  61.     public void setNextTime(Long nextTime) {
  62.         this.nextTime = nextTime;
  63.     }
  64.  
  65.     public ProducerRecord parse() {     // 解析成ProducerRecord
  66.         Integer partition = null;
  67.         Long timestamp = System.currentTimeMillis();
  68.         List<Header> headers = new ArrayList<>();
  69.         ByteBuffer retryTimesBuffer = ByteBuffer.allocate(4);
  70.         retryTimesBuffer.putInt(retryTimes);
  71.         retryTimesBuffer.flip();
  72.         headers.add(new RecordHeader(RetryRecord.KEY_RETRY_TIMES, retryTimesBuffer));
  73.  
  74.         ProducerRecord sendRecord = new ProducerRecord(
  75.                 topic, partition, timestamp, key, value, headers);
  76.         return sendRecord;
  77.     }
  78. }

本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号