com.google.common.collect.Maps#newConcurrentMap ( )源码实例Demo

public RemoteExecutionEventListener() {
  this.downloads = new LongAdder();
  this.downloadBytes = new LongAdder();
  this.uploads = new LongAdder();
  this.uploadBytes = new LongAdder();
  this.remoteCpuTimeMs = new LongAdder();
  this.remoteQueueTimeMs = new LongAdder();
  this.totalRemoteTimeMs = new LongAdder();
  this.totalBuildRules = new LongAdder();
  this.hasFirstRemoteActionStarted = new AtomicBoolean(false);

  localFallbackTotalExecutions = new LongAdder();
  localFallbackLocalExecutions = new LongAdder();
  localFallbackSuccessfulLocalExecutions = new LongAdder();

  this.actionStateCount = Maps.newConcurrentMap();
  for (State state : RemoteExecutionActionEvent.State.values()) {
    actionStateCount.put(state, new LongAdder());
  }
}
 

@QMapConfig(value = "config.properties", key = ADMIN_RATE_LIMIT)
private void onLoad(String config) {
    // config like this : key1:速率,key2:速率（单位秒）
    if (currentConfig.equals(config)) {
        // 如果config.properties文件改动，但本key并无变化，不允许触发本速率限制变更
        return;
    }

    final Map<String, Integer> limiter = parse(config);
    Map<String, RateLimiter> newRateLimiterMap = Maps.newConcurrentMap();
    for (Map.Entry<String, Integer> entry : limiter.entrySet()) {
        newRateLimiterMap.put(entry.getKey(), RateLimiter.create(entry.getValue()));
    }

    rateLimiterMap = newRateLimiterMap;

    currentConfig = config;
}
 

public static Map<String, List<String>> getMetrics() {
    if (metrics != null) {
        return metrics;
    }

    synchronized (metricMutex) {
        if (metrics == null) {
            if (getConfig() != null) {
                Map<String, List<String>> map = Maps.newConcurrentMap();
                getConfig().getDashboards().forEach(d -> {
                    if (d.getMetricVariables() != null) {
                        d.getMetricVariables().stream().forEach(v -> {
                            String delimiter = getDelimiter(v.getTarget());
                            String key = v.getName() + delimiter + d.getUid();
                            map.put(key, v.getMetrics().stream().map(m -> m.getName()).collect(Collectors.toList()));
                        });
                    }
                });
                metrics = map;
            }
        }
        return metrics;
    }
}
 

/**
 * Converts metric map into a new map which contains string formatted metric type as key.
 *
 * @param metricMap metric map in which ControlMetricType is key
 * @return a new map in which string formatted metric type is key
 */
private Map<String, Double> convertMap(Map<ControlMetricType, Double> metricMap) {
    if (metricMap == null) {
        return ImmutableMap.of();
    }
    Map newMap = Maps.newConcurrentMap();
    metricMap.forEach((k, v) -> newMap.putIfAbsent(k.toString(), v));
    return newMap;
}
 

protected Experiment(Experiments owner, String name) {
    this.owner = owner;
    this.name = name;
    this.filter = FilterExpression.alwaysTrue();
    this.hashAttributes = EMPTY_SET;
    this.allocations = new Allocations();
    this.treatments = Maps.newConcurrentMap();
    this.overrides = Maps.newConcurrentMap();
    this.seed = (int) IdentityBuilder.seed(0).putString(name).hash();
}
 

@Test
public void testToArray() {
    ExtendedSet<TestValue> set = new ExtendedSet<>(Maps.newConcurrentMap());
    TestValue val = new TestValue("foo", 1);
    assertTrue(set.add(val));
    TestValue nextval = new TestValue("goo", 2);
    assertTrue(set.add(nextval));
    Object[] array = set.toArray();
    TestValue[] valarray = {val, nextval};
    assertArrayEquals(valarray, array);
    assertTrue(set.toArray(new TestValue[0])[0] instanceof TestValue);

}
 

@Inject
public SingularityExecutorProcessKiller(
  SingularityExecutorConfiguration configuration
) {
  this.configuration = configuration;

  this.destroyFutures = Maps.newConcurrentMap();
  this.scheduledExecutorService =
    Executors.newScheduledThreadPool(
      configuration.getKillThreads(),
      new ThreadFactoryBuilder()
        .setNameFormat("SingularityExecutorKillThread-%d")
        .build()
    );
}
 

@Override
public Map<String, Object> configMap() {
    if (configMap != null) {
        return ImmutableMap.copyOf(configMap);
    } else {
        return Maps.newConcurrentMap();
    }
}
 

public Map<String, Object> summarizeProperties(String prefix) throws Exception {
    Map<String, Object> map = Maps.newConcurrentMap();

    // 从全局变量中归类
    summarizeProperties(properties, map, prefix.endsWith(ThunderProperties.DOT) ? prefix : prefix + ThunderProperties.DOT);

    // 从局部变量中归类，如果全局变量和局部变量都存在某个属性，那么就使用局部变量，否则使用全部变量
    ThunderProperties subProperties = getSubProperties();
    summarizeProperties(subProperties, map, prefix.endsWith(ThunderProperties.DOT) ? prefix : prefix + ThunderProperties.DOT);

    return map;
}
 

/**
 * Obtains the kubernetes node IP and kubernetes network gateway IP map.
 *
 * @param nodeService       kubernetes node service
 * @param networkService    kubernetes network service
 * @return kubernetes node IP and kubernetes network gateway IP map
 */
public static Map<String, String> nodeIpGatewayIpMap(K8sNodeService nodeService,
                                                     K8sNetworkService networkService) {
    Map<String, String> ipMap = Maps.newConcurrentMap();

    nodeService.completeNodes().forEach(n -> {
        K8sNetwork network = networkService.network(n.hostname());
        if (network != null) {
            ipMap.put(n.dataIp().toString(), network.gatewayIp().toString());
        }
    });

    return ipMap;
}
 

/**
 * @param ignite2 Second node.
 * @param ignMsg IgniteMessage.
 * @throws Exception If failed.
 */
private void sendOrderedMultiThreadsWith2Node(
        final Ignite ignite2,
        final IgniteMessaging ignMsg
) throws Exception {
    final ConcurrentMap<String, List<String>> expMsg = Maps.newConcurrentMap();
    final ConcurrentMap<String, List<String>> actlMsg = Maps.newConcurrentMap();

    final List<String> msgs = orderedMessages();

    sendOrderedMultiThreadsWith2Node(ignite2, ignMsg, expMsg, actlMsg, msgs);

}
 

public void addTag(String key, String value) {
    if(tags == null) {
        synchronized (this) {
            if(tags == null) {
                tags = Maps.newConcurrentMap();
            }
        }
    }
    tags.put(key, value);
}
 

private ClientMgr() {
	nameToCluster = Maps.newHashMap();
	remoteMap = Maps.newConcurrentMap();
}
 

@Override
public ServerServiceDefinition export(Class<?> protocol, Object protocolImpl) {
  Class<?> serivce = protocol;
  Object serviceRef = protocolImpl;
  String serviceName = protocol.getName();
  ServerServiceDefinition.Builder serviceDefBuilder =
      ServerServiceDefinition.builder(serviceName);
  List<Method> methods = ReflectUtils.findAllPublicMethods(serivce);
  if (methods.isEmpty()) {
    throw new IllegalStateException(
        "protocolClass " + serviceName + " not have export method" + serivce);
  }
  final ConcurrentMap<String, AtomicInteger> concurrents = Maps.newConcurrentMap();
  for (Method method : methods) {
    MethodDescriptor<Message, Message> methodDescriptor =
        GrpcUtil.createMethodDescriptor(serivce, method);
    GrpcMethodType grpcMethodType = method.getAnnotation(GrpcMethodType.class);
    switch (grpcMethodType.methodType()) {
      case UNARY:
        serviceDefBuilder.addMethod(methodDescriptor,
            ServerCalls.asyncUnaryCall(new ServerInvocation(serviceRef, method, grpcMethodType,
                providerUrl, concurrents, clientServerMonitor)));
        break;
      case CLIENT_STREAMING:
        serviceDefBuilder.addMethod(methodDescriptor,
            ServerCalls.asyncClientStreamingCall(new ServerInvocation(serviceRef, method,
                grpcMethodType, providerUrl, concurrents, clientServerMonitor)));
        break;
      case SERVER_STREAMING:
        serviceDefBuilder.addMethod(methodDescriptor,
            ServerCalls.asyncServerStreamingCall(new ServerInvocation(serviceRef, method,
                grpcMethodType, providerUrl, concurrents, clientServerMonitor)));
        break;
      case BIDI_STREAMING:
        serviceDefBuilder.addMethod(methodDescriptor,
            ServerCalls.asyncBidiStreamingCall(new ServerInvocation(serviceRef, method,
                grpcMethodType, providerUrl, concurrents, clientServerMonitor)));
        break;
      default:
        RpcServiceException rpcFramwork =
            new RpcServiceException(RpcErrorMsgConstant.SERVICE_UNFOUND);
        throw rpcFramwork;
    }
  }
  log.info("'{}' service has been registered.", serviceName);
  return serviceDefBuilder.build();
}
 

public InputReadyTracker() {
  readyInputs = Maps.newConcurrentMap();
}
 

private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException
{
  input.defaultReadObject();
  oldOperatorToWindowIdsMap = Maps.newConcurrentMap();
}
 

private PushDispatchMgr() {
	pushClassMap = Maps.newHashMap();
	syncRemote = Maps.newConcurrentMap();
	asyncRemote = Maps.newConcurrentMap();
	syncRemoteDiscard = Maps.newConcurrentMap();
}
 

@Override
public synchronized void init(JobEngineConfig jobEngineConfig, JobLock lock) throws SchedulerException {
    jobLock = lock;

    String serverMode = jobEngineConfig.getConfig().getServerMode();
    if (!("job".equals(serverMode.toLowerCase(Locale.ROOT)) || "all".equals(serverMode.toLowerCase(Locale.ROOT)))) {
        logger.info("server mode: " + serverMode + ", no need to run job scheduler");
        return;
    }
    logger.info("Initializing Job Engine ....");

    if (!initialized) {
        initialized = true;
    } else {
        return;
    }

    this.jobEngineConfig = jobEngineConfig;

    if (jobLock.lockJobEngine() == false) {
        throw new IllegalStateException("Cannot start job scheduler due to lack of job lock");
    }

    //load all executable, set them to a consistent status
    fetcherPool = Executors.newScheduledThreadPool(1);
    int corePoolSize = jobEngineConfig.getMaxConcurrentJobLimit();
    jobPool = new ThreadPoolExecutor(corePoolSize, corePoolSize, Long.MAX_VALUE, TimeUnit.DAYS,
            new SynchronousQueue<Runnable>());
    context = new DefaultContext(Maps.<String, Executable> newConcurrentMap(), jobEngineConfig.getConfig());

    logger.info("Starting resume all running jobs.");
    ExecutableManager executableManager = getExecutableManager();
    executableManager.resumeAllRunningJobs();
    logger.info("Finishing resume all running jobs.");

    int pollSecond = jobEngineConfig.getPollIntervalSecond();

    logger.info("Fetching jobs every {} seconds", pollSecond);
    JobExecutor jobExecutor = new JobExecutor() {
        @Override
        public void execute(AbstractExecutable executable) {
            jobPool.execute(new JobRunner(executable));
        }
    };
    fetcher = jobEngineConfig.getJobPriorityConsidered()
            ? new PriorityFetcherRunner(jobEngineConfig, context, jobExecutor)
            : new DefaultFetcherRunner(jobEngineConfig, context, jobExecutor);
    logger.info("Creating fetcher pool instance:" + System.identityHashCode(fetcher));
    fetcherPool.scheduleAtFixedRate(fetcher, pollSecond / 10, pollSecond, TimeUnit.SECONDS);
    hasStarted = true;
}
 

private void setGroupBuckets(Service service, boolean install) {
    Map<ServicePort, Set<String>> spEpasMap = getSportEpAddressMap(service);
    Map<ServicePort, List<GroupBucket>> spGrpBkts = Maps.newConcurrentMap();
    Map<String, String> nodeIpGatewayIpMap =
            nodeIpGatewayIpMap(k8sNodeService, k8sNetworkService);

    for (K8sNode node : k8sNodeService.completeNodes()) {
        spEpasMap.forEach((sp, epas) -> {
            List<GroupBucket> bkts = Lists.newArrayList();

            for (String ip : epas) {
                GroupBucket bkt = buildBuckets(node.intgBridge(),
                        nodeIpGatewayIpMap.getOrDefault(ip, ip), sp);

                if (bkt == null) {
                    continue;
                }

                if (install) {
                    bkts.add(bkt);
                } else {
                    bkts.remove(bkt);
                }
            }

            spGrpBkts.put(sp, bkts);
        });

        String serviceIp = service.getSpec().getClusterIP();
        spGrpBkts.forEach((sp, bkts) -> {
            String svcStr = servicePortStr(serviceIp, sp.getPort(), sp.getProtocol());
            int groupId = svcStr.hashCode();

            if (bkts.size() > 0) {
                k8sGroupRuleService.setBuckets(appId, node.intgBridge(), groupId, bkts);
            }
        });

        spEpasMap.forEach((sp, epas) ->
                // add flow rules for unshifting IP domain
                epas.forEach(epa -> {

                    String podIp = nodeIpGatewayIpMap.getOrDefault(epa, epa);

                    int targetPort;
                    if (sp.getTargetPort().getIntVal() == null) {
                        Pod pod = podByIp(k8sPodService, podIp);
                        targetPort = portNumberByName(pod, sp.getTargetPort().getStrVal());
                    } else {
                        targetPort = sp.getTargetPort().getIntVal();
                    }

                    if (targetPort != 0) {
                        setUnshiftDomainRules(node.intgBridge(), POD_TABLE,
                                PRIORITY_NAT_RULE, serviceIp, sp.getPort(),
                                sp.getProtocol(), podIp,
                                targetPort, install);
                    }
                })
        );
    }
}
 

/**
 * Should be called once execution is over, and the intra-build cache of in-progress computations
 * should be discarded. If the cache is non-empty (due to an interrupted/failed build), failure to
 * call complete() can both cause a memory leak and incorrect results on the subsequent build.
 */
public void complete(ExtendedEventHandler eventHandler) {
  // Discard all remaining state (there should be none after a successful execution).
  stateMap = Maps.newConcurrentMap();
  actionRewindStrategy.reset(eventHandler);
}