集合的分类集合List和Set都继承于Collection，都是单个储存元素的都是可以迭代的。1. List特点是是可重复、有序的，存入和迭代取出的次序相同。List的子类又有ArrayList、LinkedList、Vector。ArrayList底层采用数组结构存储，适合做查询笔试和频繁的增删，单线程操作效率高，随机增删效率低，多线程不安全。 初始10，按照1.5倍扩容Vector和ArrayList底层原理相同，但是使用sychorinized保证了多线程安全，效率低少使用。 初始10，按照2倍扩容LinkedList底层采用双向链表结构存储元素，增删效率高，随机查找效率低。不能通过数学表达式计算被查找元素的内存地址，每一次查找都是从头节点开始遍历，直到找到为止。所以LinkedList集合检索/查找的效率较低。2. Set特点是元素无序不可重复，不能按照index取值，适合做数据字典。HashSet底层是一个哈希表(散列表)，实际是一个HashMap。SortedSet特点是存取无序且不可重复，但是存的元素可以自动按照大小排列。TreeSet继承于SortedSet，底层是一个二叉树TreeMap。3. Map双列集合Map，和Collection集合没有关系，Map集合以key-value的形式储存元素，key和value都是存储java对象的内存地址，所有Map集合的特点都是无序不可重复。Map和Set集合存储元素的特点相同。HashMap底层是哈希表，非线程安全，在JDK8后如果哈希表单向元素超过8个就编程红黑树结构，红黑树的节点数量小于6个又编程单项链表结构，为的是提交检索效率。 初始容量是16，按照2倍扩容。HashTable底层也是哈希表，用sychorinized是线程安全的，key和value不允许为null。 初始容量11，按照 原容量*2+1 方式扩容。Properties继承于HashTable，线程安全，key和value只能是String类型。常用方法1.Collection的常用方法方法名说明boolean add(E e)添加元素到集合的末尾（追加）boolean remove(Object o)删除指定的元素，成功则返回true（底层调用equles）void clear()清空集合boolean contains(Object o)判断元素在集合中是否存在，存在则返回true（底层调用equles）boolean isEmpty()判断集合是否为空，空则返回trueint size()返回集合中元素个数Iterator iterator()迭代器2.ArrayList的常用方法方法名说明public ArrayList()创建一个空集合public boolean add(E e)将指定的参数元素追加到集合的末尾public void add(int index ,E e)在集合的指定位置添加指定的元素（插入元素）public void addAll(E object)用于将指定集合中所有元素添加到当前集合中public boolean remove(Object o)删除指定的元素，成功则返回truepublic E remove(int index)删除指定索引位置的元素，返回被删除的元素public E set(int index,E e)修改指定索引位置的元素，返回修改前的元素public E get(int index)获取指定索引对应的元素public int size()获取结合中元素个数3.Map的基本方法方法名说明V put(K key,V value)设置键值对V remove(Object key)删除元素void clear()清空集合boolean containsKey(Object key)判断键是否存在，存在则返回trueboolean containsValue(Object value)判断值是否存在，存在则返回trueboolean isEmpty()判断集合是否为空int size()获取集合元素个数

为什么我提交了好几次，但是系统显示我的排名和分数一直不变

想问问，Java版本调试的方法，有没有具体教程，上面的说明不是看的很懂

【如题】Java该怎么安排文件路劲呢？放src下的.java 类都被删掉了。总不能一个Main写几千行吧 XD

 1.青蛙过河 1.题目描述 小青蛙住在一条河边, 它想到河对岸的学校去学习。小青蛙打算经过河里 的石头跳到对岸。  河里的石头排成了一条直线, 小青蛙每次跳跃必须落在一块石头或者岸上。 不过, 每块石头有一个高度, 每次小青蛙从一块石头起跳, 这块石头的高度就 会下降 1 , 当石头的高度下降到 0 时小青蛙不能再跳到这块石头上（某次跳跃 后使石头高度下降到 0 是允许的）。  小青蛙一共需要去学校上 x xx 天课, 所以它需要往返 2 x 2x2x 次。当小青蛙具有 一个跳跃能力 y yy 时, 它能跳不超过 y yy 的距离。  请问小青蛙的跳跃能力至少是多少才能用这些石头上完 x xx 次课。  2.输入格式 输入的第一行包含两个整数 n , x n,xn,x, 分别表示河的宽度和小青蛙需要去学校 的天数。请注意 2 x 2x2x 才是实际过河的次数。 第二行包含 n − 1 n−1n−1 个非负整数 H 1 , H 2 , ⋯ , H n − 1 H_1,H_2,⋯,H_{n-1}H  1 ​  ,H  2 ​  ,⋯,H  n−1 ​  , 其中 H i > 0 H_i>0H  i ​  >0表 示在河中与 小青蛙的家相距 i ii 的地方有一块高度为 H i H_iH  i ​   ​的石头, H i = 0 H_i =0H  i ​  =0 表示这个位置没有石头。  3.输出格式 输出一行, 包含一个整数, 表示小青蛙需要的最低跳跃能力。  4.样例输入 5 1 1 0 1 0  5.样例输出 4  6.数据范围 1 ≤ n ≤ 1 0 5 , 1 ≤ x ≤ 1 0 9 , 1 ≤ H i ≤ 1 0 4 。 1≤n≤10^5 ,1≤x≤10^9,1≤H i ≤10^ 4 。1≤n≤10  5  ,1≤x≤10  9  ,1≤Hi≤10  4  。  7.原题链接 青蛙过河  2.解题思路 假设青蛙可以按照某条路线S SS从家跳往对岸，路线S SS上所有的石子高度均减1，这个操作等价于“青蛙从对岸按照路线S SS反向跳回家，路线S SS上所有的石子高度均减1”。  这也说明，判断小青蛙能否往返2 x 2x2x次，等价于判断小青蛙能否从左往右跳重复2 x 2x2x次。  由题目可以发现，设小青蛙的跳跃能力为y yy，当小青蛙跳跃能力y yy越大，越容易满足“重复2x次”的约束，即求解的y yy存在单调性：  当y yy越大时，小青蛙每次可以跳的范围更大，可以跳更少的步数到达对岸，即更容易重复2 x 2x2x次，当y = n y=ny=n时，无需经过任何石子就可以跳到对岸。 当y yy越小时，小青蛙需要使用更多的步数才能到达对岸，更不容易满足“重复2 x 2x2x次”的约束。 本题最终需要求解的是：恰好满足约束的最小的y yy 答案存在单调性，显然可以用二分答案的算法进行求解，初始区间[ l , r ] = [ 1 , n ] [l,r]=[1,n][l,r]=[1,n]：  求出区间[ l , r ] [l,r][l,r]的中点m i d midmid，m i d = ( l + r ) / / 2 mid=(l+r)//2mid=(l+r)//2 判断当小青蛙跳跃能力等于m i d midmid时，能否从左往右跳重复2 x 2x2x次 如果可以，则更新a n s = m i d ans=midans=mid，调整搜索区间为[ l , m i d − 1 ] [l,mid-1][l,mid−1]（求最小值，因此调整右端点） 否则，调整搜索区间为[ m i d + 1 , r ] [mid+1,r][mid+1,r] 如果l > r l > rl>r，终止循环，否则回到1 11 二分答案将求解最值问题转换成判定性问题，问题转变成当跳跃能力等于y yy时，判断小青蛙能否从左往右跳2 x 2x2x次。  小青蛙最开始位于0处，跳跃能力等于y yy，需要重复跳跃2 x 2x2x次，则首先要求从1 − y 1-y1−y的石子高度必须大于等于2 x 2x2x，不然小青蛙迈出的第一步都无法重复2 x 2x2x次。  这个结论可以推广——“所有长度为y yy的区间中石子高度之和必须大于等于2 x 2x2x”。  如果所有长度为y yy的区间中，石子高度之和等于2 x : 2x:2x:则存在H i = H i + y H_i=H_{i+y}H  i ​  =H  i+y ​  ，则只要保证第一步在[ 1 , y ] [1,y][1,y]中选择一个可以跳跃的石子i ii，则后续跳跃只需从当前位置i ii跳到i + y i+yi+y即可。这样可以保证重复2 x 2x2x次； 如果所有长度为y yy的区间中，石子高度之和大于2 x 2x2x：**则可以考虑去除某些石子的高度，从而构造出情况1，此时也是可以保证重复2 x 2x2x次的； 如果可以重复跳跃2 x 2x2x次，所有区间长度为y yy的区间中石子高度之和大于等于2 x 2x2x：对于任意区间[ i , i + y ] [i,i+y][i,i+y]，每次跳跃必须在区间中落脚。利用反证法，如果不在区间[ i , i + y ] [i,i+y][i,i+y]中落脚，等价于从i ii的左边跳到了i + y i+yi+y的右边，此时跳跃长度超过了能力上限y yy，因此不合法。也就是说，每次跳跃对于任意长度等于y yy的区间都落脚1次，重复2 x 2x2x次则说明该区间石子之和大于等于2 x 2x2x。 通过上面三点可以证明：“当跳跃能力等于y yy时重复2 x 2x2x次”等价于“所有区间长度等于y yy的区间石子高度之和大于等于2 x 2x2x”，利用这个结论进行二分答案的判定即可。  实现过程中事先预处理前缀和，从而可以O ( 1 )​O(1)​O(1)​求解区间和，时间复杂度O ( n log ⁡ n )​O(n\log n)​O(nlogn)​。  实际上使用双指针，维护区间和始终大于 2 x 2x2x，得到的最小区间长度则是答案，这样可做到 O ( n ) O(n)O(n) 的复杂度。  Ac_code 1.C++ #include<bits/stdc++.h> using namespace std; typedef long long LL; typedef unsigned long long uLL; typedef pair<int, int> PII; #define pb(s) push_back(s); #define SZ(s) ((int)s.size()); #define ms(s,x) memset(s, x, sizeof(s)) #define all(s) s.begin(),s.end() const int inf = 0x3f3f3f3f; const int mod = 1000000007; const int N = 200010;  LL n, x; void solve() {     cin >> n >> x;     std::vector<LL> s(n + 1);     for (int i = 1; i < n; ++i) {         cin >> s[i];         s[i] += s[i - 1];     }     //最后一块石头，也就是终点，可以无限跳     s[n] = 1e18;     int l = 1, r = n;     auto check = [&](int g) {         for (int i = 0; i + g <= n; ++i) {             int r = i + g;             if (s[r] - s[i] < 2 * x) return false;         }         return true;     };     while (l < r) {         int mid = l + r  >> 1;         if (check(mid)) r = mid;         else l = mid + 1;     }     cout << r << '\n'; } int main() {     ios_base :: sync_with_stdio(false);     cin.tie(0); cout.tie(0);     int t = 1;     while (t--)     {         solve();     }     return 0; } 2.Java import java.util.Scanner;   public class Main {     public static void main(String[] args) {         Scanner sc=new Scanner(System.in);         int n=sc.nextInt();         long x=sc.nextLong();         long []arr=new long[n+1];         for (int i=1;i<n;i++){             arr[i]=sc.nextLong()+arr[i-1];         }         arr[n]=100000000000L;         int l=0;         int ans=0;         for (int r=1;r<=n;r++){             if (arr[r]-arr[l]>= 2*x){                 ans=Math.max(ans,r-l);                 l+=1;             }         }         System.out.println(ans);     } } ———————————————— 版权声明：本文为CSDN博主「执 梗」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。 原文链接：https://blog.csdn.net/m0_57487901/article/details/129136195 

// 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 类型，也可以是数组或者不定长参数；