public class JConsumerSubscribe extends Thread { 
    public static void main(String[] args) { 
        JConsumerSubscribe jconsumer = new JConsumerSubscribe(); 
        jconsumer.start(); 
    } 
 
    /** 初始化Kafka集群信息. */ 
    private Properties configure() { 
        Properties props = new Properties(); 
        props.put("bootstrap.servers", "dn1:9092,dn2:9092,dn3:9092");// 指定Kafka集群地址 
        props.put("group.id", "ke");// 指定消费者组 
        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"); 
        return props; 
    } 
 
    /** 实现一个单线程消费者. */ 
    @Override 
    public void run() { 
        // 创建一个消费者实例对象 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configure()); 
        // 订阅消费主题集合 
        consumer.subscribe(Arrays.asList("test_kafka_topic")); 
        // 实时消费标识 
        boolean flag = true; 
        while (flag) { 
            // 获取主题消息数据 
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); 
            for (ConsumerRecord<String, String> record : records) 
                // 循环打印消息记录 
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); 
        } 
        // 出现异常关闭消费者对象 
        consumer.close(); 
    } 
}

private ConsumerRecords<K, V> poll(final long timeoutMs, final boolean includeMetadataInTimeout) { 
        acquireAndEnsureOpen(); 
        try { 
            if (timeoutMs < 0) throw new IllegalArgumentException("Timeout must not be negative"); 
 
            if (this.subscriptions.hasNoSubscriptionOrUserAssignment()) { 
                throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions"); 
            } 
 
            // poll for new data until the timeout expires 
            long elapsedTime = 0L; 
            do { 
 
                client.maybeTriggerWakeup(); 
 
                final long metadataEnd; 
                if (includeMetadataInTimeout) { 
                    final long metadataStart = time.milliseconds(); 
                    if (!updateAssignmentMetadataIfNeeded(remainingTimeAtLeastZero(timeoutMs, elapsedTime))) { 
                        return ConsumerRecords.empty(); 
                    } 
                    metadataEnd = time.milliseconds(); 
                    elapsedTime += metadataEnd - metadataStart; 
                } else { 
                    while (!updateAssignmentMetadataIfNeeded(Long.MAX_VALUE)) { 
                        log.warn("Still waiting for metadata"); 
                    } 
                    metadataEnd = time.milliseconds(); 
                } 
 
                final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollForFetches(remainingTimeAtLeastZero(timeoutMs, elapsedTime)); 
 
                if (!records.isEmpty()) { 
                    // before returning the fetched records, we can send off the next round of fetches 
                    // and avoid block waiting for their responses to enable pipelining while the user 
                    // is handling the fetched records. 
                    // 
                    // NOTE: since the consumed position has already been updated, we must not allow 
                    // wakeups or any other errors to be triggered prior to returning the fetched records. 
                    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(); 
        } 
    }

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(); 
 
        // We do not want to be stuck blocking in poll if we are missing some positions 
        // since the offset lookup may be backing off after a failure 
 
        // NOTE: the use of cachedSubscriptionHashAllFetchPositions means we MUST call 
        // updateAssignmentMetadataIfNeeded before this method. 
        if (!cachedSubscriptionHashAllFetchPositions && pollTimeout > retryBackoffMs) { 
            pollTimeout = retryBackoffMs; 
        } 
 
        client.poll(pollTimeout, startMs, () -> { 
            // since a fetch might be completed by the background thread, we need this poll condition 
            // to ensure that we do not block unnecessarily in poll() 
            return !fetcher.hasCompletedFetches(); 
        }); 
 
        // after the long poll, we should check whether the group needs to rebalance 
        // prior to returning data so that the group can stabilize faster 
        if (coordinator.rejoinNeededOrPending()) { 
            return Collections.emptyMap(); 
        } 
 
        return fetcher.fetchedRecords(); 
    }

提示： 
这里需要注意的是，max.poll.records返回的是一个poll请求的数据总和，与多少个分区无关。因此，每次消费从所有分区中拉取Topic的数据的总条数不会超过max.poll.records所设置的值。

    private final int sessionTimeoutMs; 
    private final int heartbeatIntervalMs; 
    private final int maxPollIntervalMs; 
    private final long retryBackoffMs; 
    private volatile long lastHeartbeatSend;  
    private long lastHeartbeatReceive; 
    private long lastSessionReset; 
    private long lastPoll; 
    private boolean heartbeatFailed;

public void run() { 
try { 
log.debug("Heartbeat thread started"); 
while (true) { 
synchronized (AbstractCoordinator.this) { 
if (closed) 
return; 
if (!enabled) { 
AbstractCoordinator.this.wait(); 
continue; 
} 
if (state != MemberState.STABLE) { 
// the group is not stable (perhaps because we left the group or because the coordinator 
// kicked us out), so disable heartbeats and wait for the main thread to rejoin. 
                            disable(); 
continue; 
} 
client.pollNoWakeup(); 
long now = time.milliseconds(); 
if (coordinatorUnknown()) { 
if (findCoordinatorFuture != null || lookupCoordinator().failed()) 
// the immediate future check ensures that we backoff properly in the case that no 
// brokers are available to connect to. 
AbstractCoordinator.this.wait(retryBackoffMs); 
} else if (heartbeat.sessionTimeoutExpired(now)) { 
// the session timeout has expired without seeing a successful heartbeat, so we should 
// probably make sure the coordinator is still healthy. 
                            markCoordinatorUnknown(); 
} else if (heartbeat.pollTimeoutExpired(now)) { 
// the poll timeout has expired, which means that the foreground thread has stalled 
// in between calls to poll(), so we explicitly leave the group. 
                            maybeLeaveGroup(); 
} else if (!heartbeat.shouldHeartbeat(now)) { 
// poll again after waiting for the retry backoff in case the heartbeat failed or the 
// coordinator disconnected 
AbstractCoordinator.this.wait(retryBackoffMs); 
} else { 
heartbeat.sentHeartbeat(now); 
sendHeartbeatRequest().addListener(new RequestFutureListener<Void>() { 
@Override 
public void onSuccess(Void value) { 
synchronized (AbstractCoordinator.this) { 
heartbeat.receiveHeartbeat(time.milliseconds()); 
} 
} 
@Override 
public void onFailure(RuntimeException e) { 
synchronized (AbstractCoordinator.this) { 
if (e instanceof RebalanceInProgressException) { 
// it is valid to continue heartbeating while the group is rebalancing. This 
// ensures that the coordinator keeps the member in the group for as long 
// as the duration of the rebalance timeout. If we stop sending heartbeats, 
// however, then the session timeout may expire before we can rejoin. 
heartbeat.receiveHeartbeat(time.milliseconds()); 
} else { 
heartbeat.failHeartbeat(); 
// wake up the thread if it's sleeping to reschedule the heartbeat 
                                            AbstractCoordinator.this.notify(); 
} 
} 
} 
}); 
} 
} 
} 
} catch (AuthenticationException e) { 
log.error("An authentication error occurred in the heartbeat thread", e); 
this.failed.set(e); 
} catch (GroupAuthorizationException e) { 
log.error("A group authorization error occurred in the heartbeat thread", e); 
this.failed.set(e); 
} catch (InterruptedException | InterruptException e) { 
Thread.interrupted(); 
log.error("Unexpected interrupt received in heartbeat thread", e); 
this.failed.set(new RuntimeException(e)); 
} catch (Throwable e) { 
log.error("Heartbeat thread failed due to unexpected error", e); 
if (e instanceof RuntimeException) 
this.failed.set((RuntimeException) e); 
else 
this.failed.set(new RuntimeException(e)); 
} finally { 
log.debug("Heartbeat thread has closed"); 
} 
}

View Code

public boolean sessionTimeoutExpired(long now) { 
return now - Math.max(lastSessionReset, lastHeartbeatReceive) > sessionTimeoutMs; 
} 
public boolean pollTimeoutExpired(long now) { 
return now - lastPoll > maxPollIntervalMs;
}