最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

AsyncHttpClient?ChannelPool線程池頻道池源碼流程解析

 更新時間:2023年12月08日 09:55:08   作者:codecraft  
這篇文章主要為大家介紹了AsyncHttpClient ChannelPool線程池頻道池源碼流程解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪

本文主要研究一下AsyncHttpClient的ChannelPool

ChannelPool

org/asynchttpclient/channel/ChannelPool.java

public interface ChannelPool {
  /**
   * Add a channel to the pool
   *
   * @param channel      an I/O channel
   * @param partitionKey a key used to retrieve the cached channel
   * @return true if added.
   */
  boolean offer(Channel channel, Object partitionKey);
  /**
   * Remove the channel associated with the uri.
   *
   * @param partitionKey the partition used when invoking offer
   * @return the channel associated with the uri
   */
  Channel poll(Object partitionKey);
  /**
   * Remove all channels from the cache. A channel might have been associated
   * with several uri.
   *
   * @param channel a channel
   * @return the true if the channel has been removed
   */
  boolean removeAll(Channel channel);
  /**
   * Return true if a channel can be cached. A implementation can decide based
   * on some rules to allow caching Calling this method is equivalent of
   * checking the returned value of {@link ChannelPool#offer(Channel, Object)}
   *
   * @return true if a channel can be cached.
   */
  boolean isOpen();
  /**
   * Destroy all channels that has been cached by this instance.
   */
  void destroy();
  /**
   * Flush partitions based on a predicate
   *
   * @param predicate the predicate
   */
  void flushPartitions(Predicate<Object> predicate);
  /**
   * @return The number of idle channels per host.
   */
  Map<String, Long> getIdleChannelCountPerHost();
}
ChannelPool定義了offer、poll、removeAll、isOpen、destroy、flushPartitions、getIdleChannelCountPerHost方法,它有兩個實現類,分別是NoopChannelPool及DefaultChannelPool

NoopChannelPool

org/asynchttpclient/channel/NoopChannelPool.java

public enum NoopChannelPool implements ChannelPool {
  INSTANCE;
  @Override
  public boolean offer(Channel channel, Object partitionKey) {
    return false;
  }
  @Override
  public Channel poll(Object partitionKey) {
    return null;
  }
  @Override
  public boolean removeAll(Channel channel) {
    return false;
  }
  @Override
  public boolean isOpen() {
    return true;
  }
  @Override
  public void destroy() {
  }
  @Override
  public void flushPartitions(Predicate<Object> predicate) {
  }
  @Override
  public Map<String, Long> getIdleChannelCountPerHost() {
    return Collections.emptyMap();
  }
}
NoopChannelPool是個枚舉,用枚舉實現了單例,其方法默認為空操作

DefaultChannelPool

/**
 * A simple implementation of {@link ChannelPool} based on a {@link java.util.concurrent.ConcurrentHashMap}
 */
public final class DefaultChannelPool implements ChannelPool {
  private static final Logger LOGGER = LoggerFactory.getLogger(DefaultChannelPool.class);
  private final ConcurrentHashMap<Object, ConcurrentLinkedDeque<IdleChannel>> partitions = new ConcurrentHashMap<>();
  private final ConcurrentHashMap<ChannelId, ChannelCreation> channelId2Creation;
  private final AtomicBoolean isClosed = new AtomicBoolean(false);
  private final Timer nettyTimer;
  private final int connectionTtl;
  private final boolean connectionTtlEnabled;
  private final int maxIdleTime;
  private final boolean maxIdleTimeEnabled;
  private final long cleanerPeriod;
  private final PoolLeaseStrategy poolLeaseStrategy;
  public DefaultChannelPool(AsyncHttpClientConfig config, Timer hashedWheelTimer) {
    this(config.getPooledConnectionIdleTimeout(),
            config.getConnectionTtl(),
            hashedWheelTimer,
            config.getConnectionPoolCleanerPeriod());
  }
  public DefaultChannelPool(int maxIdleTime,
                            int connectionTtl,
                            Timer nettyTimer,
                            int cleanerPeriod) {
    this(maxIdleTime,
            connectionTtl,
            PoolLeaseStrategy.LIFO,
            nettyTimer,
            cleanerPeriod);
  }
  public DefaultChannelPool(int maxIdleTime,
                            int connectionTtl,
                            PoolLeaseStrategy poolLeaseStrategy,
                            Timer nettyTimer,
                            int cleanerPeriod) {
    this.maxIdleTime = maxIdleTime;
    this.connectionTtl = connectionTtl;
    connectionTtlEnabled = connectionTtl > 0;
    channelId2Creation = connectionTtlEnabled ? new ConcurrentHashMap<>() : null;
    this.nettyTimer = nettyTimer;
    maxIdleTimeEnabled = maxIdleTime > 0;
    this.poolLeaseStrategy = poolLeaseStrategy;
    this.cleanerPeriod = Math.min(cleanerPeriod, Math.min(connectionTtlEnabled ? connectionTtl : Integer.MAX_VALUE, maxIdleTimeEnabled ? maxIdleTime : Integer.MAX_VALUE));
    if (connectionTtlEnabled || maxIdleTimeEnabled)
      scheduleNewIdleChannelDetector(new IdleChannelDetector());
  }
  //......
}
DefaultChannelPool基于ConcurrentHashMap實現了ChannelPool接口,主要的參數為connectionTtl、maxIdleTime、cleanerPeriod、poolLeaseStrategy;cleanerPeriod會取connectionTtl、maxIdleTime、傳入的cleanerPeriod的最小值;開啟connectionTtl或者maxIdleTime的話,會往nettyTimer添加IdleChannelDetector,延后cleanerPeriod時間執(zhí)行

offer

public boolean offer(Channel channel, Object partitionKey) {
    if (isClosed.get())
      return false;
    long now = unpreciseMillisTime();
    if (isTtlExpired(channel, now))
      return false;
    boolean offered = offer0(channel, partitionKey, now);
    if (connectionTtlEnabled && offered) {
      registerChannelCreation(channel, partitionKey, now);
    }
    return offered;
  }
  private boolean isTtlExpired(Channel channel, long now) {
    if (!connectionTtlEnabled)
      return false;
    ChannelCreation creation = channelId2Creation.get(channel.id());
    return creation != null && now - creation.creationTime >= connectionTtl;
  }  
  private boolean offer0(Channel channel, Object partitionKey, long now) {
    ConcurrentLinkedDeque<IdleChannel> partition = partitions.get(partitionKey);
    if (partition == null) {
      partition = partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>());
    }
    return partition.offerFirst(new IdleChannel(channel, now));
  }  
  private void registerChannelCreation(Channel channel, Object partitionKey, long now) {
    ChannelId id = channel.id();
    if (!channelId2Creation.containsKey(id)) {
      channelId2Creation.putIfAbsent(id, new ChannelCreation(now, partitionKey));
    }
  }
offer接口先判斷isTtlExpired,如果channel的存活時間超過connectionTtl則返回false,否則執(zhí)行offer0,往ConcurrentLinkedDeque<IdleChannel>添加,若添加成功且connectionTtlEnabled則執(zhí)行registerChannelCreation,維護創(chuàng)建時間

poll

/**
   * {@inheritDoc}
   */
  public Channel poll(Object partitionKey) {
    IdleChannel idleChannel = null;
    ConcurrentLinkedDeque<IdleChannel> partition = partitions.get(partitionKey);
    if (partition != null) {
      while (idleChannel == null) {
        idleChannel = poolLeaseStrategy.lease(partition);
        if (idleChannel == null)
          // pool is empty
          break;
        else if (!Channels.isChannelActive(idleChannel.channel)) {
          idleChannel = null;
          LOGGER.trace("Channel is inactive, probably remotely closed!");
        } else if (!idleChannel.takeOwnership()) {
          idleChannel = null;
          LOGGER.trace("Couldn't take ownership of channel, probably in the process of being expired!");
        }
      }
    }
    return idleChannel != null ? idleChannel.channel : null;
  }
poll方法是根據partitionKey找到對應的ConcurrentLinkedDeque<IdleChannel>,然后循環(huán)執(zhí)行poolLeaseStrategy.lease(partition),若idleChannel為null直接break,若isChannelActive為false則重置為null繼續(xù)循環(huán),若idleChannel.takeOwnership()為false也重置為null繼續(xù)循環(huán)

removeAll

/**
   * {@inheritDoc}
   */
  public boolean removeAll(Channel channel) {
    ChannelCreation creation = connectionTtlEnabled ? channelId2Creation.remove(channel.id()) : null;
    return !isClosed.get() && creation != null && partitions.get(creation.partitionKey).remove(new IdleChannel(channel, Long.MIN_VALUE));
  }
removeAll方法會將指定的channel從channelId2Creation及ConcurrentLinkedDeque<IdleChannel>中移除

isOpen

/**
   * {@inheritDoc}
   */
  public boolean isOpen() {
    return !isClosed.get();
  }
isOpen則取的isClosed變量

destroy

/**
   * {@inheritDoc}
   */
  public void destroy() {
    if (isClosed.getAndSet(true))
      return;
    partitions.clear();
    if (connectionTtlEnabled) {
      channelId2Creation.clear();
    }
  }
destroy會設置isClosed為true,然后清空partitions及channelId2Creation

flushPartitions

public void flushPartitions(Predicate<Object> predicate) {
    for (Map.Entry<Object, ConcurrentLinkedDeque<IdleChannel>> partitionsEntry : partitions.entrySet()) {
      Object partitionKey = partitionsEntry.getKey();
      if (predicate.test(partitionKey))
        flushPartition(partitionKey, partitionsEntry.getValue());
    }
  }
  private void flushPartition(Object partitionKey, ConcurrentLinkedDeque<IdleChannel> partition) {
    if (partition != null) {
      partitions.remove(partitionKey);
      for (IdleChannel idleChannel : partition)
        close(idleChannel.channel);
    }
  }
  private void close(Channel channel) {
    // FIXME pity to have to do this here
    Channels.setDiscard(channel);
    if (connectionTtlEnabled) {
      channelId2Creation.remove(channel.id());
    }
    Channels.silentlyCloseChannel(channel);
  }
flushPartitions會遍歷partitions,然后執(zhí)行predicate.test,為true則執(zhí)行flushPartition,它將從partitions移除指定的partitionKey,然后遍歷idleChannels挨個執(zhí)行close

getIdleChannelCountPerHost

public Map<String, Long> getIdleChannelCountPerHost() {
    return partitions
            .values()
            .stream()
            .flatMap(ConcurrentLinkedDeque::stream)
            .map(idle -> idle.getChannel().remoteAddress())
            .filter(a -> a.getClass() == InetSocketAddress.class)
            .map(a -> (InetSocketAddress) a)
            .map(InetSocketAddress::getHostName)
            .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
  }
getIdleChannelCountPerHost則遍歷partitions,然后map出remoteAddress獲取hostName,然后進行groupBy

PoolLeaseStrategy

public enum PoolLeaseStrategy {
    LIFO {
      public <E> E lease(Deque<E> d) {
        return d.pollFirst();
      }
    },
    FIFO {
      public <E> E lease(Deque<E> d) {
        return d.pollLast();
      }
    };
    abstract <E> E lease(Deque<E> d);
  }
PoolLeaseStrategy是個枚舉,定義了LIFO及FIFO兩個枚舉,LIFO則是對Deque執(zhí)行pollFirst,FIFO則是對Deque執(zhí)行pollLast

IdleChannelDetector

private final class IdleChannelDetector implements TimerTask {
    private boolean isIdleTimeoutExpired(IdleChannel idleChannel, long now) {
      return maxIdleTimeEnabled && now - idleChannel.start >= maxIdleTime;
    }
    private List<IdleChannel> expiredChannels(ConcurrentLinkedDeque<IdleChannel> partition, long now) {
      // lazy create
      List<IdleChannel> idleTimeoutChannels = null;
      for (IdleChannel idleChannel : partition) {
        boolean isIdleTimeoutExpired = isIdleTimeoutExpired(idleChannel, now);
        boolean isRemotelyClosed = !Channels.isChannelActive(idleChannel.channel);
        boolean isTtlExpired = isTtlExpired(idleChannel.channel, now);
        if (isIdleTimeoutExpired || isRemotelyClosed || isTtlExpired) {
          LOGGER.debug("Adding Candidate expired Channel {} isIdleTimeoutExpired={} isRemotelyClosed={} isTtlExpired={}", idleChannel.channel, isIdleTimeoutExpired, isRemotelyClosed, isTtlExpired);
          if (idleTimeoutChannels == null)
            idleTimeoutChannels = new ArrayList<>(1);
          idleTimeoutChannels.add(idleChannel);
        }
      }
      return idleTimeoutChannels != null ? idleTimeoutChannels : Collections.emptyList();
    }
    private List<IdleChannel> closeChannels(List<IdleChannel> candidates) {
      // lazy create, only if we hit a non-closeable channel
      List<IdleChannel> closedChannels = null;
      for (int i = 0; i < candidates.size(); i++) {
        // We call takeOwnership here to avoid closing a channel that has just been taken out
        // of the pool, otherwise we risk closing an active connection.
        IdleChannel idleChannel = candidates.get(i);
        if (idleChannel.takeOwnership()) {
          LOGGER.debug("Closing Idle Channel {}", idleChannel.channel);
          close(idleChannel.channel);
          if (closedChannels != null) {
            closedChannels.add(idleChannel);
          }
        } else if (closedChannels == null) {
          // first non closeable to be skipped, copy all
          // previously skipped closeable channels
          closedChannels = new ArrayList<>(candidates.size());
          for (int j = 0; j < i; j++)
            closedChannels.add(candidates.get(j));
        }
      }
      return closedChannels != null ? closedChannels : candidates;
    }
    public void run(Timeout timeout) {
      if (isClosed.get())
        return;
      if (LOGGER.isDebugEnabled())
        for (Object key : partitions.keySet()) {
          int size = partitions.get(key).size();
          if (size > 0) {
            LOGGER.debug("Entry count for : {} : {}", key, size);
          }
        }
      long start = unpreciseMillisTime();
      int closedCount = 0;
      int totalCount = 0;
      for (ConcurrentLinkedDeque<IdleChannel> partition : partitions.values()) {
        // store in intermediate unsynchronized lists to minimize
        // the impact on the ConcurrentLinkedDeque
        if (LOGGER.isDebugEnabled())
          totalCount += partition.size();
        List<IdleChannel> closedChannels = closeChannels(expiredChannels(partition, start));
        if (!closedChannels.isEmpty()) {
          if (connectionTtlEnabled) {
            for (IdleChannel closedChannel : closedChannels)
              channelId2Creation.remove(closedChannel.channel.id());
          }
          partition.removeAll(closedChannels);
          closedCount += closedChannels.size();
        }
      }
      if (LOGGER.isDebugEnabled()) {
        long duration = unpreciseMillisTime() - start;
        if (closedCount > 0) {
          LOGGER.debug("Closed {} connections out of {} in {} ms", closedCount, totalCount, duration);
        }
      }
      scheduleNewIdleChannelDetector(timeout.task());
    }
  }
IdleChannelDetector實現了netty的TimerTask接口,其run方法主要是遍歷partitions,通過expiredChannels取出過期的IdleChannel,這里isIdleTimeoutExpired、isRemotelyClosed、isTtlExpired都算在內,然后挨個執(zhí)行takeOwnership及close,再從channelId2Creation及partition中移除,最后再次調度一下IdleChannelDetector

小結

AsyncHttpClient的ChannelPool定義了offer、poll、removeAll、isOpen、destroy、flushPartitions、getIdleChannelCountPerHost方法,它有兩個實現類,分別是NoopChannelPool及DefaultChannelPool;DefaultChannelPool基于ConcurrentHashMap實現了ChannelPool接口,主要的參數為connectionTtl、maxIdleTime、cleanerPeriod、poolLeaseStrategy;cleanerPeriod會取connectionTtl、maxIdleTime、傳入的cleanerPeriod的最小值;開啟connectionTtl或者maxIdleTime的話,會往nettyTimer添加IdleChannelDetector,延后cleanerPeriod時間執(zhí)行。

poll方法會判斷是active,不是的話繼續(xù)循環(huán)lease,而IdleChannelDetector則會定期檢查,isIdleTimeoutExpired、isRemotelyClosed、isTtlExpired都會被close,offer的時候還會判斷isTtlExpired,這樣子來保證連接的活性。

以上就是AsyncHttpClient ChannelPool的詳細內容,更多關于AsyncHttpClient ChannelPool的資料請關注腳本之家其它相關文章!

相關文章

  • Java根據實體生成SQL數據庫表的示例代碼

    Java根據實體生成SQL數據庫表的示例代碼

    這篇文章主要來和大家分享一個Java實現根據實體生成SQL數據庫表的代碼,文中的實現代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下
    2023-07-07
  • Java中Springboot集成Kafka實現消息發(fā)送和接收功能

    Java中Springboot集成Kafka實現消息發(fā)送和接收功能

    Kafka是一個高吞吐量的分布式發(fā)布-訂閱消息系統(tǒng),主要用于處理大規(guī)模數據流,它由生產者、消費者、主題、分區(qū)和代理等組件構成,Kafka可以實現消息隊列、數據存儲和流處理等功能,在Java中,可以使用Spring Boot集成Kafka實現消息的發(fā)送和接收,感興趣的朋友跟隨小編一起看看吧
    2025-01-01
  • Spring Data Jpa 復合主鍵的實現

    Spring Data Jpa 復合主鍵的實現

    這篇文章主要介紹了Spring Data Jpa 復合主鍵的實現,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
    2019-04-04
  • 讓你五分鐘徹底理解Spring MVC

    讓你五分鐘徹底理解Spring MVC

    其實MVC就是處理Web請求的一種框架模式,如果你對MVC不太熟悉的話可以看下本文,這篇文章主要給大家介紹了關于如何讓你五分鐘徹底理解Spring MVC的相關資料,需要的朋友可以參考下
    2021-10-10
  • java密鑰交換算法DH定義與應用實例分析

    java密鑰交換算法DH定義與應用實例分析

    這篇文章主要介紹了java密鑰交換算法DH定義與應用,結合實例形式分析了Java密鑰交換算法DH的原理、定義、使用方法及相關操作注意事項,需要的朋友可以參考下
    2019-09-09
  • Java Socket編程實例(四)- NIO TCP實踐

    Java Socket編程實例(四)- NIO TCP實踐

    這篇文章主要講解Java Socket編程中NIO TCP的實例,希望能給大家做一個參考。
    2016-06-06
  • Java內存溢出實現原因及解決方案

    Java內存溢出實現原因及解決方案

    這篇文章主要介紹了Java內存溢出實現原因及解決方案,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
    2020-03-03
  • SpringBoot中的Thymeleaf模板

    SpringBoot中的Thymeleaf模板

    Thymeleaf 的出現是為了取代 JSP,雖然 JSP 存在了很長時間,并在 Java Web 開發(fā)中無處不在,但是它也存在一些缺陷。在這篇文中給大家介紹了這些缺陷所存在問題,對spring boot thymeleaf 模板相關知識感興趣的朋友跟隨小編一起看看吧
    2018-10-10
  • Java操作MongoDB模糊查詢和分頁查詢

    Java操作MongoDB模糊查詢和分頁查詢

    這篇文章主要介紹了Java操作MongoDB模糊查詢和分頁查詢的相關資料,需要的朋友可以參考下
    2016-04-04
  • LinkedList學習示例模擬堆棧與隊列數據結構

    LinkedList學習示例模擬堆棧與隊列數據結構

    這篇文章主要介紹了LinkedList學習示例,模擬一個堆棧與隊列數據結構,大家參考使用吧
    2014-01-01

最新評論

桐庐县| 金阳县| 万载县| 苏尼特右旗| 阿坝| 荆州市| 中阳县| 蚌埠市| 湘潭市| 星座| 教育| 白沙| 日土县| 大余县| 修武县| 利川市| 淅川县| 乌拉特后旗| 崇仁县| 蛟河市| 仁布县| 西平县| 香河县| 陕西省| 开江县| 双辽市| 宣汉县| 七台河市| 灵寿县| 岳阳市| 德清县| 阿克苏市| 潮州市| 玉龙| 靖宇县| 灵武市| 东阿县| 余庆县| 安康市| 兴安盟| 金华市|