Kafka之消费与激情(5)

                    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();  

（编辑：ASP站长网）