Kafka之消费与激情

                    if (fetcher.sendFetches() > 0 || client.hasPendingRequests()) {  

                        client.pollNoWakeup();                    }                    return this.interceptors.onConsume(new ConsumerRecords<>(records));  

                }                final long fetchEnd = time.milliseconds();                elapsedTime += fetchEnd - metadataEnd;            } while (elapsedTime < timeoutMs);  

            return ConsumerRecords.empty();  

        } finally {  

            release();        }    }  

上述代码中有个方法pollForFetches，它的实现逻辑如下：

private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollForFetches(final long timeoutMs) {  

        final long startMs = time.milliseconds();  

        long pollTimeout = Math.min(coordinator.timeToNextPoll(startMs), timeoutMs);  

        // if data is available already, return it immediately  

        final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchedRecords();  

        if (!records.isEmpty()) {  

            return records;  

        }  

        // send any new fetches (won't resend pending fetches)  

        fetcher.sendFetches();  

（编辑：辽源站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!