面试官问我：线程锁导致的kafka客户端超时，如何解决？

try {

// 这里发出了请求

client.send(request, time.milliseconds());

while (client.active()) {

List<ClientResponse> responses = client.poll(Long.MAX_VALUE, time.milliseconds());

for (ClientResponse response : responses) {

if (response.requestHeader().correlationId() == request.correlationId()) {

if (response.wasDisconnected()) {

throw new IOException("Connection to " + response.destination() + " was disconnected before the response was read");

}

if (response.versionMismatch() != null) {

throw response.versionMismatch();

}

return response;

}

}

}

throw new IOException("Client was shutdown before response was read");

} catch (DisconnectException e) {

if (client.active())

throw e;

else

throw new IOException("Client was shutdown before response was read");

}

@Override

public List<ClientResponse> poll(long timeout, long now) {

ensureActive();

if (!abortedSends.isEmpty()) {

// If there are aborted sends because of unsupported version exceptions or disconnects,

// handle them immediately without waiting for Selector#poll.

List<ClientResponse> responses = new ArrayList<>();

handleAbortedSends(responses);

completeResponses(responses);

return responses;

}

long metadataTimeout = metadataUpdater.maybeUpdate(now);

try {

this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));

} catch (IOException e) {

log.error("Unexpected error during I/O", e);

}

// process completed actions

long updatedNow = this.time.milliseconds();

List<ClientResponse> responses = new ArrayList<>();

handleCompletedSends(responses, updatedNow);

handleCompletedReceives(responses, updatedNow);

handleDisconnections(responses, updatedNow);

handleConnections();

handleInitiateApiVersionRequests(updatedNow);

// 关键的超时判断

handleTimedOutRequests(responses, updatedNow);

completeResponses(responses);

return responses;

}

"metrics-1@746" prio=5 tid=0xf nid=NA runnable

java.lang.Thread.State: RUNNABLE

at sun.management.ThreadImpl.getThreadInfo(Native Method)

at sun.management.ThreadImpl.getThreadInfo(ThreadImpl.java:185)

at sun.management.ThreadImpl.getThreadInfo(ThreadImpl.java:149)

import java.lang.management.ManagementFactory;

import java.lang.management.ThreadInfo;

import java.lang.management.ThreadMXBean;

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

public class ThreadLockSimple {

public static void main(String[] args) {

for (int i = 0; i < 15_000; i++) {

new Thread(new Runnable() {

@Override

public void run() {

try {

Thread.sleep(200_000);

} catch (InterruptedException e) {

throw new RuntimeException(e);

}

}

}).start();

}

ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

executorService.scheduleAtFixedRate(new Runnable() {

@Override

public void run() {

System.out.println("take " + " " + System.currentTimeMillis());

}

}, 1, 1, TimeUnit.SECONDS);

ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();

ScheduledExecutorService metricsService = Executors.newSingleThreadScheduledExecutor();

metricsService.scheduleAtFixedRate(new Runnable() {

@Override

public void run() {

long start = System.currentTimeMillis();

ThreadInfo[] threadInfoList = threadMXBean.getThreadInfo(threadMXBean.getAllThreadIds());

System.out.println("threads count " + threadInfoList.length + " cost :" + (System.currentTimeMillis() - start));

}

}, 1, 1, TimeUnit.SECONDS);

}

}