// Endpoint以北京四为例，其他地区请按实际情况填写。 String endPoint = "https://obs.cn-north-4.myhuaweicloud.com"; String ak = "*** Provide your Access Key ***"; String sk = "*** Provide your Secret Key ***"; // 创建ObsClient实例 ObsClient obsClient = new ObsClient(ak, sk, endPoint); obsClient.deleteObject("bucketname", "objectname");使用这种方式删除对象不成功，返回信息：DeleteObjectResult [deleteMarker=false, objectKey=obj-test-sg-3, versionId=null]，删除对象要如何操作？

1 使用java.util.Timer 这种方式的定时任务主要用到两个类，Timer 和 TimerTask，使用起来比较简单。其中 Timer 负责设定 TimerTask 的起始与间隔执行时间。 TimerTask是一个抽象类，new的时候实现自己的 run 方法，然后将其丢给 Timer 去执行即可。 代码示例： import java.time.LocalDateTime; import java.util.Timer; import java.util.TimerTask;  public class Schedule {     public static void main(String[] args) {         TimerTask timerTask = new TimerTask() {             @Override             public void run() {                 System.out.println("当前线程：" + Thread.currentThread().getName() + " 当前时间" + LocalDateTime.now());             }         };         // 在指定延迟0毫秒后开始，随后地执行以2000毫秒间隔执行timerTask          new Timer().schedule(timerTask, 0L, 2000L);         System.out.println("当前线程：" + Thread.currentThread().getName() + " 当前时间" + LocalDateTime.now());     } }  缺点： Timer 的背后只有一个线程，不管有多少个任务，都只有一个工作线程串行执行，效率低下 受限于单线程，如果第一个任务逻辑上死循环了，后续的任务一个都得不到执行 依然是由于单线程，任一任务抛出异常后，整个 Timer 就会结束，后续任务全部都无法执行 2 使用ScheduledExecutorService ScheduledExecutorService 即是 Timer 的替代者，JDK 1.5 并发包引入，是基于线程池设计的定时任务类。每个调度任务都会分配到线程池中的某一个线程去执行，任务就是并发调度执行的，任务之间互不影响。  Java 5.0引入了java.util.concurrent包，其中的并发实用程序之一是ScheduledThreadPoolExecutor ，它是一个线程池，用于以给定的速率或延迟重复执行任务。它实际上是Timer/TimerTask组合的更通用替代品，因为它允许多个服务线程，接受各种时间单位，并且不需要子类TimerTask （只需实现Runnable）。使用一个线程配置ScheduledThreadPoolExecutor使其等效于Timer 。 代码示例： import java.time.LocalDateTime; import java.util.concurrent.*;  public class Schedule {     public static void main(String[] args) {         // 创建一个ScheduledThreadPoolExecutor线程池，核心线程数为5         ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(5);         // 创建Runnable打印当前线程和当前时间         Runnable r = () -> System.out.println("当前线程：" + Thread.currentThread().getName() + " 当前时间" + LocalDateTime.now());         /**          * schedule：只执行一次调度          * scheduleAtFixedRate：一开始就计算间隔时间，如果任务超过间隔时间，那么就直接开始下一个任务          * scheduleWithFixedDelay：任务无论执行多久，都要等待上一轮任务完成之后再间隔指定时间，然后才开始下一个任务          */          // 在指定1秒延迟后执行r,之后每两秒执行一次         scheduledExecutorService.scheduleAtFixedRate(r, 1, 2, TimeUnit.SECONDS);     } }  3 使用Spring Task Spring Task 底层是基于 JDK 的 ScheduledThreadPoolExecutor 线程池来实现的。直接通过Spring 提供的 @Scheduled 注解即可定义定时任务，非常方便。 以Spring Boot来作为示例，步骤为 在启动类所在包下创建Schedule 类（在没有配置@ComponentScan的情况下，Spring Boot只会默认扫描启动类所在包的spring组件） 在该类上添加@Component和@EnableScheduling注解 在方法上添加@Scheduled注解，该注解主要参数如下 String cron() default "";  // 支持cron表达式  long fixedDelay() default -1;  // 在最后一次调用结束和下一次调用开始之间的时间间隔，以毫秒为单位 String fixedDelayString() default "";  // 同上，类似ScheduledExecutorService的scheduleWithFixedDelay  long fixedRate() default -1;  // 在调用之前的时间间隔，以毫秒为单位 String fixedRateString() default "";  // 同上，类似ScheduledExecutorService的scheduleAtFixedRate  long initialDelay() default -1;  // 在第一次执行fixedRate()或fixedDelay()任务之前要延迟的毫秒数 String initialDelayString() default "";  // 同上 代码示例：  import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component;  import java.time.LocalDateTime; @Component @EnableScheduling public class Schedule {     @Scheduled(fixedRate = 2000L)     public void task() {         System.out.println("当前线程：" + Thread.currentThread().getName() + " 当前时间" + LocalDateTime.now());     } }  优点： 简单，轻量，支持 Cron 表达式 缺点 ：默认只支持单机，是单线程的，并且提供的功能比较单一 可以通过@EnableAsync和 @Async开启多线程 import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component;  import java.time.LocalDateTime; @Component @EnableAsync  // 开启异步多线程 @EnableScheduling public class Schedule {      @Async     @Scheduled(fixedRate = 2000L)     public void task() {         System.out.println("当前线程：" + Thread.currentThread().getName() + " 当前时间" + LocalDateTime.now());     } }  使用@EnableAsync注解后，默认情况下，Spring将搜索关联的线程池定义：上下文中的唯一org.springframework.core.task.TaskExecutor 的bean，或者名为“taskExecutor”的java.util.concurrent.Executor 的bean。如果两者都无法解析，则将使用org.springframework.core.task.SimpleAsyncTaskExecutor来处理异步方法调用。  TaskExecutor实现为每个任务启动一个新线程，异步执行它。 支持通过“concurrencyLimit”bean 属性限制并发线程。默认情况下，并发线程数是无限的，所以使用默认的线程池有导致内存溢出的风险。 注意：刚才的运行结果看起来是线程复用的，而实际上此实现不重用线程！应尽量实现一个线程池TaskExecutor ，特别是用于执行大量短期任务。不要使用默认的SimpleAsyncTaskExecutor。  import org.springframework.context.annotation.Bean; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component;  import java.time.LocalDateTime; import java.util.concurrent.Executor; @Component @EnableAsync @EnableScheduling public class Schedule {      @Async     @Scheduled(fixedRate = 2000L)     public void task() {         System.out.println("当前线程：" + Thread.currentThread().getName() + " 当前时间" + LocalDateTime.now());     }      @Bean("taskExecutor")     public Executor taskExecutor() {         ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();         taskExecutor.setCorePoolSize(10);         taskExecutor.setMaxPoolSize(50);         taskExecutor.setQueueCapacity(200);         taskExecutor.setKeepAliveSeconds(60);         taskExecutor.setThreadNamePrefix("自定义-");         taskExecutor.setAwaitTerminationSeconds(60);         return taskExecutor;     } } ———————————————— 原文链接：https://blog.csdn.net/dreaming9420/article/details/124003021 

1.消费位移确认     Kafka消费者消费位移确认有自动提交与手动提交两种策略。在创建KafkaConsumer对象时，通过参数enable.auto.commit设定，true表示自动提交（默认）。自动提交策略由消费者协调器（ConsumerCoordinator）每隔${auto.commit.interval.ms}毫秒执行一次偏移量的提交。手动提交需要由客户端自己控制偏移量的提交。         (1)自动提交。在创建一个消费者时，默认是自动提交偏移量，当然我们也可以显示设置为自动。例如，我们创建一个消费者，该消费者自动提交偏移量  Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("client.id", "test"); props.put("enable.auto.commit", true);// 显示设置偏移量自动提交 props.put("auto.commit.interval.ms", 1000);// 设置偏移量提交时间间隔 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// 创建消费者 consumer.subscribe(Arrays.asList("test"));// 订阅主题         (2)手动提交。在有些场景我们可能对消费偏移量有更精确的管理，以保证消息不被重复消费以及消息不被丢失。假设我们对拉取到的消息需要进行写入数据库处理，或者用于其他网络访问请求等等复杂的业务处理，在这种场景下，所有的业务处理完成后才认为消息被成功消费，这种场景下，我们必须手动控制偏移量的提交。          Kafka 提供了异步提交（commitAsync）及同步提交（commitSync）两种手动提交的方式。两者的主要区别在于同步模式下提交失败时一直尝试提交，直到遇到无法重试的情况下才会结束，同时，同步方式下消费者线程在拉取消息时会被阻塞，直到偏移量提交操作成功或者在提交过程中发生错误。而异步方式下消费者线程不会被阻塞，可能在提交偏移量操作的结果还未返 回时就开始进行下一次的拉取操作，在提交失败时也不会尝试提交。         实现手动提交前需要在创建消费者时关闭自动提交，即设置enable.auto.commit=false。然后在业务处理成功后调用commitAsync()或commitSync()方法手动提交偏移量。由于同步提交会阻塞线程直到提交消费偏移量执行结果返回，而异步提交并不会等消费偏移量提交成功后再继续下一次拉取消息的操作，因此异步提交还提供了一个偏移量提交回调的方法commitAsync(OffsetCommitCallback callback)。当提交偏移量完成后会回调OffsetCommitCallback 接口的onComplete()方法，这样客户端根据回调结果执行不同的逻辑处理。  Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("client.id", "test"); props.put("fetch.max.bytes", 1024);// 为了便于测试，这里设置一次fetch 请求取得的数据最大值为1KB,默认是5MB props.put("enable.auto.commit", false);// 设置手动提交偏移量 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); // 订阅主题 consumer.subscribe(Arrays.asList("test")); try {     int minCommitSize = 10;// 最少处理10 条消息后才进行提交     int icount = 0 ;// 消息计算器     while (true) {         // 等待拉取消息         ConsumerRecords<String, String> records = consumer.poll(1000);         for (ConsumerRecord<String, String> record : records) {             // 简单打印出消息内容,模拟业务处理             System.out.printf("partition = %d, offset = %d,key= %s value = %s%n", record. partition(), record.offset(), record.key(),record.value());             icount++;         }         // 在业务逻辑处理成功后提交偏移量         if (icount >= minCommitSize){             consumer.commitAsync(new OffsetCommitCallback() {                 @Override                 public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {                     if (null == exception) {                     // TODO 表示偏移量成功提交                     System.out.println("提交成功");                     } else {                         // TODO 表示提交偏移量发生了异常，根据业务进行相关处理                         System.out.println("发生了异常");                     }                 }             });             icount=0; // 重置计数器         }     } } catch(Exception e){     // TODO 异常处理     e.printStackTrace(); } finally {     consumer.close(); }     3.5以时间戳查询消息         Kafka 在0.10.1.1 版本增加了时间戳索引文件，因此我们除了直接根据偏移量索引文件查询消息之外，还可以根据时间戳来访问消息。consumer-API 提供了一个offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch)方法，该方法入参为一个Map 对象，Key 为待查询的分区，Value 为待查询的时间戳，该方法会返回时间戳大于等于待查询时间的第一条消息对应的偏移量和时间戳。需要注意的是，若待查询的分区不存在，则该方法会被一直阻塞。         假设我们希望从某个时间段开始消费，那们就可以用offsetsForTimes()方法定位到离这个时间最近的第一条消息的偏移量，在查到偏移量之后调用seek(TopicPartition partition, long offset)方法将消费偏移量重置到所查询的偏移量位置，然后调用poll()方法长轮询拉取消息。例如，我们希望从主题“stock-quotation”第0 分区距离当前时间相差12 小时之前的位置开始拉取消息 Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("client.id", "test"); props.put("enable.auto.commit", true);// 显示设置偏移量自动提交 props.put("auto.commit.interval.ms", 1000);// 设置偏移量提交时间间隔 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); // 订阅主题 consumer.assign(Arrays.asList(new TopicPartition("test", 0))); try {     Map<TopicPartition, Long> timestampsToSearch = new HashMap<TopicPartition,Long>();     // 构造待查询的分区     TopicPartition partition = new TopicPartition("stock-quotation", 0);     // 设置查询12 小时之前消息的偏移量     timestampsToSearch.put(partition, (System.currentTimeMillis() - 12 * 3600 * 1000));     // 会返回时间大于等于查找时间的第一个偏移量     Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes (timestampsToSearch);     OffsetAndTimestamp offsetTimestamp = null;     // 这里依然用for 轮询，当然由于本例是查询的一个分区，因此也可以用if 处理     for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetMap.entrySet()) {         // 若查询时间大于时间戳索引文件中最大记录索引时间，         // 此时value 为空,即待查询时间点之后没有新消息生成         offsetTimestamp = entry.getValue();         if (null != offsetTimestamp) {         // 重置消费起始偏移量         consumer.seek(partition, entry.getValue().offset());         }     }     while (true) {         // 等待拉取消息         ConsumerRecords<String, String> records = consumer.poll(1000);         for (ConsumerRecord<String, String> record : records){             // 简单打印出消息内容             System.out.printf("partition = %d, offset = %d,key= %s value = %s%n", record.partition(), record.offset(), record.key(),record.value());         }     } } catch (Exception e) {     e.printStackTrace(); } finally {     consumer.close(); }     3.6消费速度控制         提供 pause(Collection<TopicPartition> partitions)和resume(Collection<TopicPartition> partitions)方法，分别用来暂停某些分区在拉取操作时返回数据给客户端和恢复某些分区向客户端返回数据操作。通过这两个方法可以对消费速度加以控制，结合业务使用。 ———————————————— 原文链接：https://blog.csdn.net/qq_35349490/article/details/79790625 

Kafka client 消息接收的三种模式 引言 kafka的消费模式总共有3种：最多一次，最少一次，正好一次。为什么会有这3种模式，是因为客户端处理消息，提交反馈（commit）这两个动作不是原子性。  1.最多一次：客户端收到消息后，在处理消息前自动提交，这样kafka就认为consumer已经消费过了，偏移量增加。 2.最少一次：客户端收到消息，处理消息，再提交反馈。这样就可能出现消息处理完了，在提交反馈前，网络中断或者程序挂了，那么kafka认为这个消息还没有被consumer消费，产生重复消息推送。 3.正好一次：保证消息处理和提交反馈在同一个事务中，即有原子性。 本文从这几个点出发，详细阐述了如何实现以上三种方式。 1.At-most-once（最多一次） 设置enable.auto.commit为ture 设置 auto.commit.interval.ms为一个较小的时间间隔. client不要调用commitSync()，kafka在特定的时间间隔内自动提交。 示例     public void mostOnce(){         Properties props = new Properties();         props.put("bootstrap.servers", "localhost:9092");         props.put("group.id", "test-1");         props.put("enable.auto.commit", "true");         props.put("auto.commit.interval.ms", "1000");         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);         consumer.subscribe(Arrays.asList("my-topic", "bar"));         while (true) {             ConsumerRecords<String, String> records = consumer.poll(100);             for (ConsumerRecord<String, String> record : records) {                 process(record);             }         }     }  2.At-least-once（最少一次） 方法一 设置enable.auto.commit为false client调用commitSync()，增加消息偏移;     public void leastOnce(){         Properties props = new Properties();         props.put("bootstrap.servers", "localhost:9092");         props.put("group.id", "test-1");         props.put("enable.auto.commit", "false"); //取消自动提交         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);         consumer.subscribe(Arrays.asList("my-topic", "bar"));         while (true) {             ConsumerRecords<String, String> records = consumer.poll(100);             for (ConsumerRecord<String, String> record : records)                 process(record);                 consumer.commitAsync(); //提交offset         }     } 方法二 设置enable.auto.commit为ture 设置 auto.commit.interval.ms为一个较大的时间间隔. client调用commitSync(),增加消息偏移; 示例     public void leastOnce(){         Properties props = new Properties();         props.put("bootstrap.servers", "10.242.1.219:9092");         props.put("group.id", "test-1");         props.put("enable.auto.commit", "true"); //自动提交         props.put("auto.commit.interval.ms", "99999999");         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);         consumer.subscribe(Arrays.asList("my-topic", "bar"));         while (true) {             ConsumerRecords<String, String> records = consumer.poll(100);             for (ConsumerRecord<String, String> record : records)                 process(record);                 consumer.commitAsync(); //提交offset         }     } 3.Exactly-once（正好一次） 3.1 思路 如果要实现这种方式，必须自己控制消息的offset，自己记录一下当前的offset，对消息的处理和offset的移动必须保持在同一个事务中，例如在同一个事务中，把消息处理的结果存到mysql数据库同时更新此时的消息的偏移。 3.2 实现 设置enable.auto.commit为false 保存ConsumerRecord中的offset到数据库 当partition分区发生变化的时候需要rebalance，有以下几个事件会触发分区变化 1 consumer订阅的topic中的分区大小发生变化 2 topic被创建或者被删除 3 consuer所在group中有个成员挂了 4 新的consumer通过调用join加入了group 此时 consumer通过实现ConsumerRebalanceListener接口，捕捉这些事件，对偏移量进行处理。 consumer通过调用seek(TopicPartition, long)方法，移动到指定的分区的偏移位置。 3.接下去需要实现ConsumerRebalanceListener接口，在分区rebalance的时候，调用的顺序：先调用onPartitionsRevoked（通知consumer 任务被取消了），再调用onPartitionsAssigned（通知consumer新的任务来了）。 4.那么在我们收到任务被取消的时候，把对应offset保存到数据库；在收到新任务到来的时候，从数据库读出对应分区的偏移(例如刚启动)，具体实现如下所示。 public class MyConsumerRebalancerListener implements org.apache.kafka.clients.consumer.ConsumerRebalanceListener {      private Consumer<String, String> consumer;     private MessageDao messageDao;      public MyConsumerRebalancerListener(Consumer<String, String> consumer, MessageDao messageDao) {         this.consumer = consumer;         this.messageDao = messageDao;     }      //任务被取消     public void onPartitionsRevoked(Collection<TopicPartition> partitions) {         for (TopicPartition partition : partitions) {             long offset = consumer.position(partition);             MessageOffsetPO build = MessageOffsetPO                     .builder()                     .offset(offset)                     .kPartition(partition.partition() + "")                     .topic(partition.topic())                     .build();             try {                 messageDao.insertWinner(build);             } catch (Exception e) {              }             log.info("onPartitionsRevoked topic:{},build:{}",partition.topic(),build);         }     }      //收到新任务     public void onPartitionsAssigned(Collection<TopicPartition> partitions) {         for (TopicPartition partition : partitions) {             MessageOffsetPO messageOffsetPO = messageDao.get(partition.topic(), partition.partition() + "");             if(messageOffsetPO==null){ //接收到新的topic，即数据库中记录不存在                 consumer.seek(partition,0);             }else{                 consumer.seek(partition,messageOffsetPO.getOffset()+1);//下一个offset，所以需要加1             }             log.info("onPartitionsAssigned topic:{},messageOffsetPO:{},offset:{}",partition,messageOffsetPO);         }     } } 对于client端的代码可以这么写     /**      * 正好一次      */     public void exactlyOnce(){         Properties props = new Properties();         props.put("bootstrap.servers", "localhost:9092");         props.put("group.id", "test-1");         props.put("enable.auto.commit", "false"); //取消自动提交         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);         MyConsumerRebalancerListener rebalancerListener = new MyConsumerRebalancerListener(consumer,messageDao);         consumer.subscribe(Arrays.asList("test-new-topic-1", "new-topic"),rebalancerListener);         while (true) {             ConsumerRecords<String, String> records = consumer.poll(100);             for (ConsumerRecord<String, String> record : records) {                 boolean isException=false;                 try {                     log.info("consume record:{}",record);                     processService.process(record);                 } catch (Exception e) {                     e.printStackTrace();                     isException=true;                 }                 if(isException){                     //处理发生异常，说明数据没有被消费，为了保证能被消费，需要移动到该位置继续进行处理                     TopicPartition topicPartition=new TopicPartition(record.topic(),record.partition());                     consumer.seek(topicPartition,record.offset());                     log.info("consume exception offset:{}",record.offset());                     break;                 } //                  rebalancerListener.getOffsetMananger().saveOffsetInExternalStore(record.topic(),record.kPartition(),record.offset());             }         }     } 第19行 **processService.process(record);**对消息进行消费，同时记录了消息的偏移位置，ProcessService代码如下 @Service @Slf4j public class ProcessService {     @Autowired     MessageDao messageDao;     @Transactional(rollbackFor = Exception.class)     public void process(ConsumerRecord<String, String> record){          log.info(" record:{}",record);          //对消息进行处理，这里只是简单的打印了一下          System.out.println(">>>>>>>>>"+Thread.currentThread().getName()+"_"+record);          //更新偏移量          messageDao.update(record.offset(),record.topic(),record.partition()+"");     } } 3.4 运行程序. 场景1: consumer订阅topic——test-new-topic-1（之前未订阅过），然后producer向consumer发送两条信息,consumer收到信息，未抛出异常。得到日志如下： 2018-01-11 14:36:07,948 main INFO  MyConsumerRebalancerListener.onPartitionsAssigned:53 -onPartitionsAssigned topic:test-new-topic-1-0,messageOffsetPO:null,offset:{} 2018-01-11 14:37:15,156 main INFO  KafkaConsumeTest.exactlyOnce:106 -consume record:ConsumerRecord(topic = test-new-topic-1, partition = 0, offset = 0, CreateTime = 1515652635056, serialized key size = 1, serialized value size = 14, headers = RecordHeaders(headers = [], isReadOnly = false), key = 0, value = message:0-test) 2018-01-11 14:37:15,183 main INFO  ProcessService.process:23 - record:ConsumerRecord(topic = test-new-topic-1, partition = 0, offset = 0, CreateTime = 1515652635056, serialized key size = 1, serialized value size = 14, headers = RecordHeaders(headers = [], isReadOnly = false), key = 0, value = message:0-test) 2018-01-11 14:37:15,238 main INFO  KafkaConsumeTest.exactlyOnce:106 -consume record:ConsumerRecord(topic = test-new-topic-1, partition = 0, offset = 1, CreateTime = 1515652635062, serialized key size = 1, serialized value size = 14, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1, value = message:1-test) 2018-01-11 14:37:15,239 main INFO  ProcessService.process:23 - record:ConsumerRecord(topic = test-new-topic-1, partition = 0, offset = 1, CreateTime = 1515652635062, serialized key size = 1, serialized value size = 14, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1, value = message:1-test) 第1行，consumer收到新的任务指派，由于数据库中没有这个topic，所以获得的messgeOffsetPO为null，那么consumer需要从offset为0处开始接受数据。 第2-3行，consumer获得了数据并对数据进行处理，消息的offset=0 第4-5行，consumer获得了数据并对数据进行处理，消息的offset=1 此时数据库的状态为: 场景2:producer向consumer发送两条信息,consumer收到信息，进行处理，但抛出异常。     @Transactional(rollbackFor = Exception.class)     public void process(ConsumerRecord<String, String> record){          log.info(" record:{}",record);          System.out.println(">>>>>>>>>"+Thread.currentThread().getName()+"_"+record);          messageDao.update(record.offset(),record.topic(),record.partition()+"");          throw new RuntimeException("error");//抛出异常     } 运行后得到如下结果： 2018-01-11 14:56:48,357 main INFO  KafkaConsumeTest.exactlyOnce:116 -consume exception offset:2 2018-01-11 14:56:48,866 main INFO  KafkaConsumeTest.exactlyOnce:106 -consume record:ConsumerRecord(topic = test-new-topic-1, partition = 0, offset = 2, CreateTime = 1515653807138, serialized key size = 1, serialized value size = 14, headers = RecordHeaders(headers = [], isReadOnly = false), key = 0, value = message:0-test) 2018-01-11 14:56:48,867 main INFO  ProcessService.process:23 - record:ConsumerRecord(topic = test-new-topic-1, partition = 0, offset = 2, CreateTime = 1515653807138, serialized key size = 1, serialized value size = 14, headers = RecordHeaders(headers = [], isReadOnly = false), key = 0, value = message:0-test) ....... 可以看到，consumer一直在消费offset为2的数据，由于处理时时异常状态，所以一直在消费2，数据库此时的状态与之前的一致，offset为1，符合预期。 4.引用： kafkaClinet官方文档 kafkaProducer官方文档 老外的博客 他是用文件读写的方式实现“正好一次”，感觉不是很好，不过对我启发比较大。 ———————————————— 原文链接：https://blog.csdn.net/laojiaqi/article/details/79034798 

 1，subscribe方式：当主题分区数量变化或者consumer数量变化时，会进行rebalance；注册rebalance监听器，可以手动管理offset不注册监听器，kafka自动管理assign方式：手动将consumer与partition进行对应，kafka不会进行rebanlance 关键配置及含义 enable.auto.commit 是否自动提交自己的offset值；默认值时true auto.commit.interval.ms 自动提交时长间隔；默认值时5000 ms  consumer.commitSync(); offset提交命令； 默认配置 采用默认配置情况下，既不能完全保证At-least-once 也不能完全保证at-most-once； 比如： 在自动提交之后，数据消费流程失败，这样就会有丢失，不能保证at-least-once； 数据消费成功，但是自动提交失败，可能会导致重复消费，这样也不能保证at-most-once； 但是将自动提交时长设置得足够小，则可以最大限度地保证at-most-once； at most onece模式 基本思想是保证每一条消息commit成功之后，再进行消费处理； 设置自动提交为false，接收到消息之后，首先commit，然后再进行消费 at least onece模式 基本思想是保证每一条消息处理成功之后，再进行commit； 设置自动提交为false；消息处理成功之后，手动进行commit； 采用这种模式时，最好保证消费操作的“幂等性”，防止重复消费； exactly onece模式 核心思想是将offset作为唯一id与消息同时处理，并且保证处理的原子性； 设置自动提交为false；消息处理成功之后再提交； 比如对于关系型数据库来说，可以将id设置为消息处理结果的唯一索引，再次处理时，如果发现该索引已经存在，那么就不处理； ———————————————— 原文链接：https://blog.csdn.net/clzzzh/article/details/121156761 

 1.Kafka是什么 简单的说，Kafka是由Linkedin开发的一个分布式的消息队列系统(Message Queue)。kafka的架构师jay kreps非常喜欢franz kafka,觉得kafka这个名字很酷，因此将linkedin的消息传递系统命名为完全不相干的kafka，没有特别含义。 2.解决什么问题 kafka开发的主要初衷目标是构建一个用来处理海量日志，用户行为和网站运营统计等的数据处理框架。在结合了数据挖掘，行为分析，运营监控等需求的情况下，需要能够满足各种实时在线和批量离线处理应用场合对低延迟和批量吞吐性能的要求。从需求的根本上来说，高吞吐率是第一要求，其次是实时性和持久性。 既有的消息队列框架或者对消息传送的可靠性提供了较高的保证，由此带来较大的负担，不能满足海量高吞吐率的要求；或者完全面向实时消息处理系统，对于批量离线处理的场合无法提供足够的缓存和持久性要求。 而多数针对大数据开发应用的日志收集处理系统则通常更适合批量离线处理场合，对实时在线处理的场合支持不够。 总体而言，kafka试图提供一个同时满足在线和离线处理海量数据的消息派发系统。 3.怎么实现 kafka的集群由多个Broker服务器组成，每个类型的消息被定义为topic，同一topic内部的消息按照一定的key和算法被分区(partition)存储在不同的Broker上，消息生产者producer和消费者consumer可以在多个Broker上生产/消费topic 核心思想:以高效率作为第一设计原则，kafka的结构设计在很多方面都做了激进的取舍。 (1)极简的数据结构、应用模式 消息队列是以log文件的形式存储，消息生产者只能将消息添加到既有的文件尾部，没有任何ID信息用于消息的定位，完全依靠文件内的位移，因此消息的使用者只能依靠文件位移顺序读取消息，这样也就不需要维护复杂的支持随即读取的索引结构。 kafka broker完全不维护和协调多用户使用消息的行为模式，用户自己维护位移用来索引消息。 最小的并发访问单位就是partition分区，同一用户组内的所有用户(可以理解为同一个应用的所有并发进程)只能有一个访问同一分区，同时分区的个数是固定的，不支持动态调整。这样最大简化了多进程/分布式client之间对消息处理访问的并发控制的复杂度，当然也带来一定的使用模式上的限制(比如最大并发度完全取决于预先规划的partition的个数) 此外分区也带来一个问题就是消息只是分区内部有序而不是全局有序的。如果需要全局有序，应用需要自己靠别的机制来保证。 使用Pull模式派发消息，消息的使用情况，比如是否还有consumer没有读取，是否重复读取(改进中)等，在Broker端也完全不跟踪维护，消息的过期处理简单的由定时器定时删除(比如保留7天)，由此简化各种消息跟踪维护的开销。 (2)最大化数据传输效率 生产者和消费者可以批量读写消息减少RPC开销；使用Zero Copy方式在内核层直接将文件内容传送给网络Socket，避免应用层数据拷贝；使用合理的压缩格式等 (3)激进的内存管理模式 kafka不在JVM进程内部维护消息Cache，消息直接从文件中读写，完全依赖操作系统在文件系统层面的cache，避免在JVM中管理Cache带来的额外数据结构开销和GC带来的性能代价。基于批量处理和顺序读写的应用模式，最大化利用文件系统的Cache机制和规避文件读写相对内存读写的性能代价。 (3)HA kafka在0.8之前message是没有备份容错机制的，producer的工作模式是fire and forget，如果一个broker失效，那么相关topic分区的相关消息也就丢失了。这种设计的原因在于最初的应用模式，如日志/用户行为等消息的处理，对数据的健壮性方面要求不高，可以容忍部分数据的缺失。采用fire and forget 模式，不需要等待Broker ack，有利于提高producer的吞吐率。 不过在0.8版本中，添加了数据replica的机制，一个消息分区和多个replica分布在不同的Broker上，由leader replica负责日常读写，通过zookeeper监督failover，不同的分区的leader replica均衡负载到不同的Broker上。在这种情况下，producer可以选择不等待leader replica的Ack，部分Ack，或者完全备份完毕后Ack等不同的ack机制。这三种机制，性能依次递减 (producer吞吐量降低1-3倍)，数据健壮性则依次递增。 ———————————————— 原文链接：https://blog.csdn.net/weixin_33567029/article/details/112076328 

 Kafka安装 Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息，是一种高吞吐量的分布式发布订阅消息系统。 主要包含几个组件：  Topic：消息主题，特定消息的发布接口，每个Topic都可以分成数个Partition，用于消息的并发发送。 Producer:生产者，信息的发布者，发布者可以指定数个Partition进行发布。 Consumer:消费者，信息的使用者，同一个Group的消费者数量，最好不好超过Partition的数量，对于分区的Topic，消费者使用时需要指定相应的分区号。 Broker：服务代理 ##下载kafka SpringBoot整合kafka 当前SpringBoot版本为2.0.2.RELEASE，打包工具为Maven  消费者 a. 引入Pom <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">     <modelVersion>4.0.0</modelVersion>      <groupId>com.kafkatest</groupId>     <artifactId>producer</artifactId>     <version>1.0-SNAPSHOT</version>     <name>kafka-producer</name>     <parent>         <groupId>org.springframework.boot</groupId>         <artifactId>spring-boot-starter-parent</artifactId>         <version>2.0.2.RELEASE</version>         <relativePath/>     </parent>     <properties>         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>         <java.version>1.8</java.version>         <joda-time.version>2.3</joda-time.version>     </properties>      <dependencies>          <dependency>             <groupId>org.springframework.boot</groupId>             <artifactId>spring-boot-starter</artifactId>         </dependency>          <dependency>             <groupId>org.projectlombok</groupId>             <artifactId>lombok</artifactId>             <optional>true</optional>         </dependency>         <dependency>             <groupId>org.springframework.kafka</groupId>             <artifactId>spring-kafka</artifactId>         </dependency>     </dependencies>      <build>         <plugins>             <plugin>                 <groupId>org.springframework.boot</groupId>                 <artifactId>spring-boot-maven-plugin</artifactId>             </plugin>         </plugins>     </build> </project> b.JAVA代码 @Service public class KafkaProducerTest {     @Autowired     private KafkaTemplate<String,byte[]> kafkaTemplate;     private final String topic = "byteArray_topic1";      public void sendMessage(int key,String value){         ProducerRecord<String,byte[]> record = new ProducerRecord<>(topic,                 key%3,String.valueOf(key),value.getBytes());         kafkaTemplate.send(record);     } } 配置文件（YML） spring:   kafka:       producer:         bootstrap-servers: 172.169.0.109:9092         batch-size: 16384         retries: 0         buffer-memory: 33554432         key-serializer: org.apache.kafka.common.serialization.StringSerializer         value-serializer: org.apache.kafka.common.serialization.ByteArraySerializer 这里有一个非常陷阱的问题需要特别注意：序列化类的路径是：org.apache.kafka.common.serialization.StringSerializer 而不是 org.apache.kafka.config.serialization.StringSerializer 否则会出现如下错误：  2019-01-31 11:35:14.794 [main] WARN  o.s.c.a.AnnotationConfigApplicationContext -                 Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'kafkaProducerTest': Unsatisfied dependency expressed through field 'kafkaTemplate'; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration': Unsatisfied dependency expressed through constructor parameter 0; nested exception is org.springframework.boot.context.properties.ConfigurationPropertiesBindException: Error creating bean with name 'spring.kafka-org.springframework.boot.autoconfigure.kafka.KafkaProperties': Could not bind properties to 'KafkaProperties' : prefix=spring.kafka, ignoreInvalidFields=false, ignoreUnknownFields=true; nested exception is org.springframework.boot.context.properties.bind.BindException: Failed to bind properties under 'spring.kafka.producer.key-serializer' to java.lang.Class<?> 2019-01-31 11:35:14.810 [main] ERROR o.s.b.d.LoggingFailureAnalysisReporter -                 *************************** APPLICATION FAILED TO START *************************** Description: Failed to bind properties under 'spring.kafka.producer.key-serializer' to java.lang.Class<?>:      Property: spring.kafka.producer.key-serializer     Value: org.apache.kafka.config.serialization.StringSerializer     Origin: class path resource [application.yml]:8:25     Reason: No converter found capable of converting from type [java.lang.String] to type [java.lang.Class<?>] Action: Update your application's configuration 消费者 如果不使用并发获取、批量获取消费者的代码非常简单。  a.Pom文件 <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">     <modelVersion>4.0.0</modelVersion>      <groupId>com.kafkatest</groupId>     <artifactId>consumer</artifactId>     <version>1.0-SNAPSHOT</version>     <name>kafka-consumer</name>     <parent>         <groupId>org.springframework.boot</groupId>         <artifactId>spring-boot-starter-parent</artifactId>         <version>2.0.2.RELEASE</version>         <relativePath/> <!-- lookup parent from repository -->     </parent>     <properties>         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>         <java.version>1.8</java.version>         <joda-time.version>2.3</joda-time.version>     </properties>      <dependencies>          <dependency>             <groupId>org.springframework.boot</groupId>             <artifactId>spring-boot-starter</artifactId>         </dependency>          <dependency>             <groupId>org.projectlombok</groupId>             <artifactId>lombok</artifactId>             <optional>true</optional>         </dependency>         <dependency>             <groupId>org.springframework.kafka</groupId>             <artifactId>spring-kafka</artifactId>         </dependency>     </dependencies>      <build>         <plugins>             <plugin>                 <groupId>org.springframework.boot</groupId>                 <artifactId>spring-boot-maven-plugin</artifactId>             </plugin>         </plugins>     </build> </project> b1.Java代码(无并发访问、无批量获取) @Service @Slf4j public class Listener {     private final String topic = "byteArray_topic1";      public void listen(ConsumerRecord<String, byte[]> record){         log.info("kafka的key: " + record.key());         log.info("kafka的value: " + new String(record.value()));     } } b2.配置文件 spring:   kafka:     consumer:       enable-auto-commit: true       group-id: gridMonitorGroup       auto-commit-interval: 1000       auto-offset-reset: latest       bootstrap-servers: "172.169.0.109:9092"       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer       value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer c.Java代码（并发、批量获取） Kafka消费者配置类 批量获取关键代码： ①factory.setBatchListener(true); ②propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,50); 并发获取关键代码： factory.setConcurrency(concurrency); @Configuration @EnableKafka public class KafkaConsumerConfig {     @Value("${kafka.consumer.bootstrap-servers}")     private String servers;     @Value("${kafka.consumer.enable-auto-commit}")     private boolean enableAutoCommit;     @Value("${kafka.consumer.auto-commit-interval}")     private String autoCommitInterval;     @Value("${kafka.consumer.group-id}")     private String groupId;     @Value("${kafka.consumer.auto-offset-reset}")     private String autoOffsetReset;     @Value("${kafka.consumer.concurrency}")     private int concurrency;     @Bean     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, byte[]>> kafkaListenerContainerFactory() {         ConcurrentKafkaListenerContainerFactory<String, byte[]> factory = new ConcurrentKafkaListenerContainerFactory<>();         factory.setConsumerFactory(consumerFactory());         //并发数量         factory.setConcurrency(concurrency);         //批量获取         factory.setBatchListener(true);         factory.getContainerProperties().setPollTimeout(1500);         return factory;     }      public ConsumerFactory<String, byte[]> consumerFactory() {         return new DefaultKafkaConsumerFactory<>(consumerConfigs());     }      public Map<String, Object> consumerConfigs() {         Map<String, Object> propsMap = new HashMap<>();         propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);         propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);         propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);         propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);         propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);         propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);         propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);         //最多批量获取50个         propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,50);         return propsMap;     }      @Bean     public Listener listener() {         return new Listener();     } } Kafka消费者Listener @Service @Slf4j public class Listener {     private final String topic = "byteArray_topic1";      @KafkaListener(id="myListener",             topicPartitions ={@TopicPartition(topic = topic, partitions = { "0", "1" ,"2"})})     public void listen(List<ConsumerRecord<String, byte[]>> recordList) {         recordList.forEach((record)->{             log.info("kafka的key: " + record.key());             log.info("kafka的value: " + new String(record.value()));         });     } } 配置文件 kafka:  consumer:    enable-auto-commit: true    group-id: gridMonitorGroup    auto-commit-interval: 1000    auto-offset-reset: latest    bootstrap-servers: "172.169.0.109:9092"    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer    value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer    concurrency: 3 ———————————————— 原文链接：https://blog.csdn.net/menxin_job/article/details/86712973 

1. 创建测试的私有对象/**  * @author lirong  * @desc 测试对象  * @date 2019/06/20 20:07  */ public class Person {     private int age = 5;     private String name;     private Person(){}     private String test(String name){         System.out.println("name: "+name);         return "test";     } }2. 获取私有对象中的属性和方法/**  * @author lirong  * @desc 反射获取私有属性和方法  * @date 2019/06/20 20:10  */ public class Test {     public static void main(String[] args) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException, NoSuchFieldException {         // 1. 获取class对象         Class clazz = Person.class;         // 2. 获取私有无参构造         Constructor c = clazz.getDeclaredConstructor();         // 3. 设置访问为可见         c.setAccessible(true);         // 4. 通过构造器创建实例对象         Person person = (Person) c.newInstance();                  // 根据字段名称获取class中的字段         Field age = clazz.getDeclaredField("age");         age.setAccessible(true);         System.out.println(age.getName() + " = " + age.get(person));         // 修改私有变量的默认值         age.set(person, 18);         System.out.println(age.getName() + " = " + age.get(person));         // 5. 获取所有字段         Field[] fields = clazz.getDeclaredFields();         for (Field f : fields) {             // 设置字段的可见性             f.setAccessible(true);             String name = f.getName();             Object o = f.get(person);             System.out.println(name + " - " + o);         }         // 6. 获取所有的方法         Method[] methods = clazz.getDeclaredMethods();         for (Method m : methods) {             m.setAccessible(true);             String name = m.getName();             Object invoke = m.invoke(person, "张三");             System.out.println(name + " = "+invoke);         }     } }通过反射获取私有内部类对象首先是我们的目标对象：class Out {     //目标获取Inner对象     private class Inner {         //内部类的私有成员属性         private String inner = "ccc";     } }直接列出代码public class Main {     @SuppressWarnings({ "rawtypes", "unchecked" })     public static void main(String[] args) throws Exception {         //获取外部类         Class clzz = Out.class;         //获取外部类默认无参构造方法         Constructor con = clzz.getDeclaredConstructor();         //实例一个外部类对象         Out outObj = (Out) con.newInstance();         //获取外部类内的所有内部类         Class innerClazz[] = clzz.getDeclaredClasses();         //遍历         for (Class c : innerClazz) {             //获取修饰符的整数编码             int mod = c.getModifiers();             //返回整数编码对应的修饰符的字符串对象             String modifier = Modifier.toString(mod);             //找到被private修饰的内部类             if (modifier.contains("private")) {                 //根据内部类的特性，需要由外部类来反射获取内部类的构造方法（这里获取的是内部类的默认构造方法）                 Constructor cc = c.getDeclaredConstructor(clzz);                 //由于内部类是私有的，需要强制获取构造方法的访问权限                 cc.setAccessible(true);                 //由外部类对象来反射获取内部类的对象                 Object obj=cc.newInstance(outObj);                 //获取内部类的私有成员属性inner                 Field f=c.getDeclaredField("inner");                 //获取访问权限                 f.setAccessible(true);                 //获取内部类对象obj中的私有成员属性inner的值                 System.out.println(f.get(obj));             }         }     } }输出结果：ccc原文链接：https://www.yisu.com/zixun/622118.html

本篇内容主要讲解“Springboot如何集成Kafka进行批量消费”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Springboot如何集成Kafka进行批量消费”吧!引入依赖<dependency>                 <groupId>org.springframework.kafka</groupId>                 <artifactId>spring-kafka</artifactId>                 <version>1.3.11.RELEASE</version>             </dependency>因为我的项目的 springboot 版本是 1.5.22.RELEASE，所以引的是 1.3.11.RELEASE 的包。读者可以根据下图来自行选择对应的版本。图片更新可能不及时，详情可查看spring-kafka 官方网站。注：这里有个踩坑点，如果引入包版本不对，项目启动时会抛出org.springframework.core.log.LogAccessor 异常：java.lang.ClassNotFoundException: org.springframework.core.log.LogAccessor创建配置类/**      * kafka 配置类      */     @Configuration     @EnableKafka     public class KafkaConsumerConfig {         private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(KafkaConsumerConfig.class);         @Value("${kafka.bootstrap.servers}")         private String kafkaBootstrapServers;         @Value("${kafka.group.id}")         private String kafkaGroupId;         @Value("${kafka.topic}")         private String kafkaTopic;         public static final String CONFIG_PATH = "/home/admin/xxx/BOOT-INF/classes/kafka_client_jaas.conf";         public static final String LOCATION_PATH = "/home/admin/xxx/BOOT-INF/classes/kafka.client.truststore.jks";         @Bean         public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {             ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();             factory.setConsumerFactory(consumerFactory());             // 设置并发量，小于或者等于 Topic 的分区数             factory.setConcurrency(5);             // 设置为批量监听             factory.setBatchListener(Boolean.TRUE);             factory.getContainerProperties().setPollTimeout(30000);             return factory;         }         public ConsumerFactory<String, String> consumerFactory() {             return new DefaultKafkaConsumerFactory<>(consumerConfigs());         }         public Map<String, Object> consumerConfigs() {             Map<String, Object> props = new HashMap<>();             //设置接入点，请通过控制台获取对应Topic的接入点。             props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers);             //设置SSL根证书的路径，请记得将XXX修改为自己的路径。             //与SASL路径类似，该文件也不能被打包到jar中。             System.setProperty("java.security.auth.login.config", CONFIG_PATH);             props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, LOCATION_PATH);             //根证书存储的密码，保持不变。             props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");             //接入协议，目前支持使用SASL_SSL协议接入。             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");             //SASL鉴权方式，保持不变。             props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");             // 自动提交             props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.TRUE);             //两次Poll之间的最大允许间隔。             //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance，默认30s。             props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);             //设置单次拉取的量，走公网访问时，该参数会有较大影响。             props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);             props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);             //每次Poll的最大数量。             //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。             props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);             //消息的反序列化方式。             props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");             props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");             //当前消费实例所属的消费组，请在控制台申请之后填写。             //属于同一个组的消费实例，会负载消费消息。             props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaGroupId);             //Hostname校验改成空。             props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");             return props;         }     }注：此处通过 factory.setConcurrency(5); 配置了并发量为 5 ，假设我们线上的 Topic 有 12 个分区。那么将会是 3 个线程分配到 2 个分区，2 个线程分配到 3 个分区，3 * 2 + 2 * 3 = 12。Kafka 消费者/**      * kafka 消息消费类      */     @Component     public class KafkaMessageListener {         private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageListener.class);         @KafkaListener(topics = {"${kafka.topic}"})         public void listen(List<ConsumerRecord<String, String>> recordList) {             for (ConsumerRecord<String,String> record : recordList) {                 // 打印消息的分区以及偏移量                 LOGGER.info("Kafka Consume partition:{}, offset:{}", record.partition(), record.offset());                 String value = record.value();                 System.out.println("value = " + value);                 // 处理业务逻辑 ...             }         }     }因为我在配置类中设置了批量监听，所以此处 listen 方法的入参是List：List<ConsumerRecord<String, String>>。到此，相信大家对“Springboot如何集成Kafka进行批量消费”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！原文链接：https://www.yisu.com/zixun/622113.html

添加依赖：dependencies { compile('io.springfox:springfox-swagger2:2.2.2') compile('io.springfox:springfox-swagger-ui:2.2.2') compile('org.springframework.boot:spring-boot-starter-data-jpa') compile('org.springframework.boot:spring-boot-starter-jdbc') compile('org.springframework.boot:spring-boot-starter-web') runtime('mysql:mysql-connector-java') providedRuntime('org.springframework.boot:spring-boot-starter-tomcat') testCompile('org.springframework.boot:spring-boot-starter-test') }application.properties文件添加链接串spring.datasource.url=jdbc:mysql://localhost/yfeid spring.datasource.username=root spring.datasource.password=root spring.datasource.driver-class-name=com.mysql.jdbc.Driver #update database by model spring.jpa.show-sql= true spring.jpa.properties.hibernate.hbm2ddl.auto=update spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.MySQL5Dialect spring.jpa.hibernate.naming.physical-strategy=org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl添加实体类：Userpackage com.yunfeng.TestMySQL.Model; import java.io.Serializable; import java.util.Date; import javax.persistence.Entity; import javax.persistence.GeneratedValue; import javax.persistence.GenerationType; import javax.persistence.Id; import javax.persistence.Table; import com.fasterxml.jackson.annotation.JsonFormat; import javax.persistence.*; @Entity @Table(name = "user") public class User implements Serializable { /** * @Fields serialVersionUID : TODO */ private static final long serialVersionUID = -6550777752269466791L; @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private int id; private String name; private String password; private String address; @Column(name="loginName") private String loginName; @Temporal(TemporalType.TIMESTAMP) private Date createTime; public String getName() { return name; } public void setName(String name) { this.name = name; } public String getPassword() { return password; } public void setPassword(String password) { this.password = password; } public String getAddress() { return address; } public void setAddress(String address) { this.address = address; } public String getLoginName() { return loginName; } public void setLoginName(String loginName) { this.loginName = loginName; } @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") public Date getCreateTime() { return createTime; } public void setCreateTime(Date createTime) { this.createTime = createTime; } }建立仓库类：package com.yunfeng.TestMySQL.Repository; import java.util.List; import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; import com.yunfeng.TestMySQL.Model.User; public interface UserRepository extends JpaRepository<User, Integer> { List<User> findByNameOrAddress(String name, String address); @Query("select a from User a where a.address = ?1") public User findAddress(String address); @Query("from User a where a.id = :id") public User findByUserId(@Param("id")int userId); @Query("from User a where a.id > :userId") public Page<User> findByUserIdGreaterThan(@Param("userId")int userId,Pageable pageable); }使用package com.yunfeng.TestMySQL; import Request.*; import Response.*; import java.util.List; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.domain.Page; import org.springframework.data.domain.PageRequest; import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Sort; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import io.swagger.annotations.ApiOperation; @RestController public class TestController { @ApiOperation(value = "查找用户", httpMethod = "POST") @RequestMapping("/api/Test1") public TestResponse showPerson(@RequestBody TestRequest input) { TestResponse res = new TestResponse(); res.setUserName("zzzili"); res.setUserPass("pass"); return res; } @Autowired//(required=false) private UserRepository context; @ApiOperation(value="测试mysql2_通过JPA自带方法查询",httpMethod="POST") @RequestMapping("/api/testmysql2") public List<User> TestMysql2() { List<User> list = context.findAll(); System.out.println(list); return list; } @ApiOperation(value="测试mysql3_根据自定义方法名查询",httpMethod="POST") @RequestMapping("/api/testmysql3") public List<User> TestMysql3() { List<User> list = context.findByNameOrAddress("zz","ss"); System.out.println(list); return list; } @ApiOperation(value="测试mysql4_自定义query查询",httpMethod="POST") @RequestMapping("/api/testmysql4") public User TestMysql4() { //User user = context.findEmail("zz"); User user2 = context.findByUserId(1); System.out.println(user2); return user2; } @ApiOperation(value="测试mysql5_分页查询",httpMethod="POST") @RequestMapping("/api/testmysql5") public List<User> TestMysql5() { Pageable pageable = PageRequest.of(0,3, Sort.Direction.DESC,"id"); Page<User> list = context.findByUserIdGreaterThan(1,pageable); System.out.println(list); return list.getContent(); } }在查询时，通常需要同时根据多个属性进行查询，且查询的条件也格式各样（大于某个值、在某个范围等等），Spring Data JPA 为此提供了一些表达条件查询的关键字，大致如下：And --- 等价于 SQL 中的 and 关键字，比如 findByUsernameAndPassword(String user, Striang pwd)；Or --- 等价于 SQL 中的 or 关键字，比如 findByUsernameOrAddress(String user, String addr)；Between --- 等价于 SQL 中的 between 关键字，比如 findBySalaryBetween(int max, int min)；LessThan --- 等价于 SQL 中的 "<"，比如 findBySalaryLessThan(int max)；GreaterThan --- 等价于 SQL 中的">"，比如 findBySalaryGreaterThan(int min)；IsNull --- 等价于 SQL 中的 "is null"，比如 findByUsernameIsNull()；IsNotNull --- 等价于 SQL 中的 "is not null"，比如findByUsernameIsNotNull()；NotNull --- 与 IsNotNull 等价；Like --- 等价于 SQL 中的 "like"，比如 findByUsernameLike(String user)；NotLike --- 等价于 SQL 中的 "not like"，比如 findByUsernameNotLike(String user)；OrderBy --- 等价于 SQL 中的 "order by"，比如 findByUsernameOrderBySalaryAsc(String user)；Not --- 等价于 SQL 中的 "！ ="，比如 findByUsernameNot(String user)；In --- 等价于 SQL 中的 "in"，比如 findByUsernameIn(Collection userList) ，方法的参数可以是 Collection 类型，也可以是数组或者不定长参数；NotIn --- 等价于 SQL 中的 "not in"，比如 findByUsernameNotIn(Collection userList) ，方法的参数可以是 Collection 类型，也可以是数组或者不定长参数；

各位大佬们最近有在找工作的吗？行情咋样啊，想先了解一下情况。

function working_version_num() does not exist

 在初学Java的时候，一般我们都会从基本的数据类型开始学习，而在基本数据类型中，我认为double类型是比较难理解的，并且在以后的学习或工作中，在double类型数据这遇到的坑也是极多的。例如下面的这样一个程序public static void main(String[] args) {  System.out.println(2.0-1.1);  }  很多人会认为上面的程序会打印出0.9，但实际上，它打印的却是0.8999999999999999，这是为什么呢？  首先介绍一下，十进制小数是怎么转换为二进制数的，举个例子3.75  首先取出3.75的小数部分0.75，将其乘以要转换的进制的进制数，在这里也就是2，乘以2后得到结果1.5，取1.5的整数部分作为二进制小数的小数部分的第一位，再取1.5的小数部分0.5,乘以2后得到1.0，将1.0的整数部分1作为小数部分的第二位。以此类推，直到最后得到值0或形成无限循环。3.75转换为二进制数就是11.11；  但是我们知道在Java中double类型的数据占8个字节，所以对于无限循环的二进制小数我们只能取到它的近似值，就比如说1.1。  下面自己写了一个将double类型的数转换为二进制的程序public static void main(String[] args) {  doubleToBinary(3.75000000);  doubleToBinary(173.8125);  doubleToBinary(1.10);  }/**  * 10任何次负幂都不能精确地被表示为一个长度有限的二进制数  * @param d  */public static void doubleToBinary(double d) {    // 获得double类型的整数部分  int intPart = (int) d;  String tempStr = d + "";    // 获得double类型数的字符串形式的小数部分  String decimalPartStr = tempStr.substring(tempStr.indexOf("."));  BigDecimal decimal = new BigDecimal(decimalPartStr);    // 获得小数点后面的位数  int precision = decimal.precision();    // 最终的小数部分二进制字符串  String decimalPartBinary = decimalPartToBinary(Double.parseDouble(decimalPartStr), precision);  System.out.println(Integer.toBinaryString(intPart) + "." + decimalPartBinary);  }/**  * 将小数部分转换为二进制字符串  * @param decimalPart 小数部分  * @param precision   原始数的小数部分位数  * @return*/public static String decimalPartToBinary(double decimalPart, int precision) {    // 转换为整型  long decimalPartLong = (long) (Math.pow(10, precision) * decimalPart);  String temp = "";    int i = 0;    while (precision > 0 && i   decimalPartLong = decimalPartLong * 2;  temp += (int) (decimalPartLong / Math.pow(10, precision));        //取得除第一位之后的数，并转换为字符串  String str = (long) (decimalPartLong % Math.pow(10, precision)) + "";        if (str.charAt(str.length() - 1) == '0' && str.length() != 1) {            //去掉数最后面的0  decimalPartLong = Long.parseLong(str.substring(0, str.length() - 1));  } else {  decimalPartLong = Long.parseLong(str);  }        if (decimalPartLong == 0) {            break;  }  ++i;  precision = (decimalPartLong + "").length();  }    return temp;  }  上面程序中的main方法输出的值为11.1110101101.11011.0001100110011001100110011001100110011001100110011001100110011001  通过上面的程序，我们很容易看到测试数据1.10在小数部分是无限循环的，1.10并不能精确地表示为一个double，因此它在Java中被表示为最接近它的double值。既然Java中是对double不能精确表示的数以近似值去存储的，那么在一些需要精确计算的地方就有可能出现错误，甚至产生意想不到的结果。比如说在业务中经常碰到的货币计算。那么如何解决这个问题呢？  1.使用执行精确小数运算的BigDecimalAPI，但这里要说明一点，最好(一定)要用BigDecimal(String val)构造方法，而不要使用BigDecimal(double val)，因为BigDecimal(double val)构造方法将会使用它的参数val的精确值返回一个BigDecimal，比如new BigDecimal(1.1)将会返回一个表示1.100000000000000088817841970012523233890533447265625的BigDecimal。  2.使用整数类型进行计算，比如说int或long。将2.0元转换为200分去计算，这样System.out.println((200-110)+ "分")将会打印出正确的结果90分。  总而言之呢，在需要精确计算的地方要避免使用float和double的数据。如果上面的程序有错的地方，请指出，共同学习，谢谢。  作者：Jadyn  链接：https://www.jianshu.com/p/82688c94c58c 

2023年2月论坛问题合集1.Gauss DB 查询数据库全量表的行信息问题cid:link_1华为云serverless函数的vCPU是如何配置的？cid:link_2调用CMS 接口获取技能组队列实时监控指标，无法获取数据 cid:link_3hive客户端中执行add jar提示没有权限 cid:link_4预测在线服务报408的4205错误cid:link_5登录失败，错误码：50331748cid:link_6怎么通过定制方式把高德地图的最大显示层级放大 cid:link_7刚刚绑定域名，手机访问网站就显示网站有违法和违规内容这个怎么办，这样可以通过公安备案吗？cid:link_8登录iClient客户端，使用登录中的ip及端口号+接口文档里的接口地址，访问超时cid:link_9新手，调用华为云obs的SDK代码怎么整啊，有没有大佬教一下，有偿，装的eclipsecid:link_0播放输入V9.0无法收号cid:link_10下载WeAutomate完，安装驱动时，怎么也导入不了chrome的驱动https://bbs.huaweicloud.com/forum/thread-0223111236431889032-1-1.html

一、在注册或者使用chatGPT之前要做的准备工作。1、首先要做的，是开启全局代理。2、进入chatGPT网址：  Access denied (openai.com)  cid:link_03、没有账号的朋友还得准备虚拟短信验证码的网址： https://sms-activate.org/cn​4、有账号的朋友直接登录就行了。以上就是所有的准备工作。二、chatGPT注册过程及使用过程中报错解决方法。1、Access denied，报错1020    (发生在注册时，或者之前已经注册好，而且能使用，但现在突然报这种错误)这个错误是一个比较常见的错误。有朋友说，准备注册账号，但是进入上面的chatGPT网址，报了这个错误，这是因为你没有开启全局代理就直接访问，这是不行的，这样永远访问不了。  还有朋友说，我准备注册个账号，全局代理也开了，但是还是报这个错误，这就跟你的全局代理有关系，这个时候你就需要换其它的全局代理了。还有朋友说，我买了一个账号，但是现在登录不了，显示这个错误，这还是跟你的全局代理有关系。还有朋友说，我之前用的好好的，账号也已经注册好了，但是现在突然显示这个错误，还是一样，还是跟代理有关。综上，第一个错误的解决方法是，更换全局代理，越冷门的反而越好用。2、not  available  in  your  country（发生在注册时，或者之前已经注册好，而且能使用，但现在突然报这种错误）这个错误也是一个比较常见的错误。有朋友说，自己在sign up也就是注册账号的时候，出现了这个错误，这个错误出现的原因有两个，第一个是你没有开启全局代理或者代理不行；第二个是cookie的问题，可以在浏览器设置，cookie选项中删除全部cookie尝试；第三个原因是浏览器的问题，这个时候你可以尝试用无痕打开网页；这是可行的一个方法。还有朋友说，自己之前已经注册过账号了，并且也开启全局代理了，但是这几天登录的时候，突然显示这个错误，解决方法还是跟上面一样的三种方法。还有朋友说，我是在注册过程中，已经在sms上接收到验证码了，但是，把验证码填进去，给我显 示这个报错，这是因为你之前用港澳台或国内ip登录过了，不管什么设备，请勿使用港澳台及内地ip登录，否则不进行java操作，无论怎么换站点，永远都显示not available在你浏览器的地址栏输入javascript:     冒号后面复制以下以下这一段编码window.localStorage.removeItem(Object.keys(window.localStorage).find(i=>i.startsWith(‘@@auth0spajs’)))回车清理浏览器cookie重启重启完先全局连非港澳台ip不然就换设备，代理用全局综上，第二个错误的解决方法是，更换全局代理，或者浏览器使用无痕模式，或者在你浏览器的地址栏输入javascript:     冒号后面复制以下以下这一段编码window.localStorage.removeItem(Object.keys(window.localStorage).find(i=>i.startsWith(‘@@auth0spajs’)))回车清理浏览器cookie重启重启完先全局连非港澳台代理3、chatgpt is at capacity right now（发生在注册时，或者之前已经注册好，而且能使用，但现在突然报这种错误）这个错误也是一个比较常见的错误。大家都说这个错误没办法解决，因为这是服务器本身的问题，但我想说的是，也是有解决的办法的，那就是“挤进”服务器。如何“挤进去”呢？疯狂刷新网页，你会留意到刷新的时候能出来登录界面，只要在这个时候停下来就行了。综上，此问题解决方法是，疯狂刷新网页，即可“挤进服务器”。4、报404 502   错误这个错误解决很简单，重新进一下网址就好了。5、Signup is currently unavailable, please try again later.（出现在注册账号的时候）这个错误出现的大多数原因是，网易163邮箱用不了了，大家选择QQ邮箱注册，或者微软跟谷歌账号。总结以上就是chatGPT使用及注册过程中常见的一些错误的解决方法。转载自https://www.jb51.net/article/274727.htm