Kafka生产者异步发送消息产生堵塞

public void producerSendFuture(String topic, String data) {
      logger.info("kafka异步发送topic:" + topic + "|requestMsg:" + data);
      ListenableFuture<SendResult<String, String>> future = this.kafkaTemplate.send(topic, data);
      future.addCallback(new ListenableFutureCallback<Object>() {
　　　　　　public void onSuccess(Object o) {
               SendResult<String, String> sendResult = (SendResult)o;
               logger.info("成功发送消息,"offset: " +  sendResult.getRecordMetadata().offset());
              }

          public void onFailure(Throwable throwable) {
               throwable.printStackTrace();
               logger.info("发送kafka消息失败，异常原因："+ throwable.getMessage());
          }
      });
      logger.info("return");
 }                            

public class Metadata implements Closeable {
    private final Logger log;
    // retry.backoff.ms: 默认值为100ms，它用来设定两次重试之间的时间间隔，避免无效的频繁重试。
    private final long refreshBackoffMs;
    // metadata.max.age.ms: 默认值为300000，如果在这个时间内元数据没有更新的话会被强制更新。
    private final long metadataExpireMs;
    // 更新版本号，每更新成功1次，version自增1,主要是用于判断metadata是否更新
    private int updateVersion;
    // 请求版本号，每发送一次请求，version自增1
    private int requestVersion;
    // 上一次更新的时间（包含更新失败）
    private long lastRefreshMs;
    // 上一次更新成功的时间
    private long lastSuccessfulRefreshMs;
    private KafkaException fatalException;
    // 非法的topics
    private Set<String> invalidTopics;
    // 未认证的topics
    private Set<String> unauthorizedTopics;
    // 元数据信息的Cache缓存
    private MetadataCache cache = MetadataCache.empty();
    private boolean needFullUpdate;
    private boolean needPartialUpdate;
    // 会收到metadata updates的Listener列表
    private final ClusterResourceListeners clusterResourceListeners;
    private boolean isClosed;
    // 存储Partition最近一次的leaderEpoch
    private final Map<TopicPartition, Integer> lastSeenLeaderEpochs;
}
​
//MetadataCache：Kafka 集群中关于 node、topic 和 partition 的信息
public class MetadataCache {
    private final String clusterId;
    private final Map<Integer, Node> nodes;
    private final Set<String> unauthorizedTopics;
    private final Set<String> invalidTopics;
    private final Set<String> internalTopics;
    private final Node controller;
    private final Map<TopicPartition, PartitionMetadata> metadataByPartition;
    private Cluster clusterInstance;
}
​
//关于 topic 的详细信息（leader 所在节点、replica 所在节点、isr 列表）都是在 Cluster 实例中保存的
public final class Cluster {
    private final boolean isBootstrapConfigured;
    // node 列表
    private final List<Node> nodes;
    // 未认证的topics
    private final Set<String> unauthorizedTopics;
    // 非法的topics
    private final Set<String> invalidTopics;
    // kafka内置的topics
    private final Set<String> internalTopics;
    private final Node controller;
    // partition对应的信息，如：leader所在节点、所有的副本、ISR中的副本、offline的副本
    private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition;
    // topic和partition信息的对应关系
    private final Map<String, List<PartitionInfo>> partitionsByTopic;
    // topic和可用partition(leader不为null)的对应关系
    private final Map<String, List<PartitionInfo>> availablePartitionsByTopic;
    // node和partition信息的对应关系
    private final Map<Integer, List<PartitionInfo>> partitionsByNode;
    // 节点id与节点的对应关系
    private final Map<Integer, Node> nodesById;
    // 集群信息，里面只有一个clusterId
    private final ClusterResource clusterResource;
}