com.google.common.collect.Multimaps#synchronizedSetMultimap ( )源码实例Demo

/**
 * Constructor.
 */
public RemoteConfigLongPollService() {
  m_longPollFailSchedulePolicyInSecond = new ExponentialSchedulePolicy(1, 120); //in second
  m_longPollingStopped = new AtomicBoolean(false);
  m_longPollingService = Executors.newSingleThreadExecutor(
      ApolloThreadFactory.create("RemoteConfigLongPollService", true));
  m_longPollStarted = new AtomicBoolean(false);
  m_longPollNamespaces =
      Multimaps.synchronizedSetMultimap(HashMultimap.<String, RemoteConfigRepository>create());
  m_notifications = Maps.newConcurrentMap();
  m_remoteNotificationMessages = Maps.newConcurrentMap();
  m_responseType = new TypeToken<List<ApolloConfigNotification>>() {
  }.getType();
  gson = new Gson();
  m_configUtil = ApolloInjector.getInstance(ConfigUtil.class);
  m_httpUtil = ApolloInjector.getInstance(HttpUtil.class);
  m_serviceLocator = ApolloInjector.getInstance(ConfigServiceLocator.class);
  m_longPollRateLimiter = RateLimiter.create(m_configUtil.getLongPollQPS());
}
 

private Consultant(ScheduledExecutorService executor, ObjectMapper mapper, URI consulUri,
		ServiceIdentifier identifier, SetMultimap<String, SettingListener> settingListeners,
		Set<ConfigListener> configListeners, ConfigValidator validator, CloseableHttpClient http,
		boolean pullConfig, String healthEndpoint, String kvPrefix, long whenLocatingServicesCacheResultsFor) {

	this.registered = new AtomicBoolean();
	this.settingListeners = Multimaps.synchronizedSetMultimap(settingListeners);
	this.configListeners = Sets.newConcurrentHashSet(configListeners);
	this.serviceInstanceBackend = new ServiceInstanceBackend(identifier.getDatacenter(), consulUri, mapper, http,
			whenLocatingServicesCacheResultsFor);

	this.mapper = mapper;
	this.validator = validator;
	this.executor = executor;
	this.consulUri = consulUri;
	this.id = identifier;
	this.pullConfig = pullConfig;
	this.validated = new Properties();
	this.healthEndpoint = healthEndpoint;
	this.http = http;
	this.kvPrefix = kvPrefix;
}
 

@Test
default void storeShouldKeepConsistencyWhenConcurrentStore() throws Exception {
    EventDeadLetters eventDeadLetters = eventDeadLetters();

    ImmutableMap<Integer, Group> groups = concurrentGroups();
    Multimap<Integer, EventDeadLetters.InsertionId> storedInsertionIds = Multimaps.synchronizedSetMultimap(HashMultimap.create());

    ConcurrentTestRunner.builder()
        .operation((threadNumber, step) -> {
            Event.EventId eventId = Event.EventId.random();
            EventDeadLetters.InsertionId insertionId = eventDeadLetters.store(groups.get(threadNumber), event(eventId)).block();
            storedInsertionIds.put(threadNumber, insertionId);
        })
        .threadCount(THREAD_COUNT)
        .operationCount(OPERATION_COUNT)
        .runSuccessfullyWithin(RUN_SUCCESSFULLY_IN);

    groups.forEach((groupId, group) -> {
        Group storedGroup = groups.get(groupId);
        assertThat(eventDeadLetters.failedIds(storedGroup).collectList().block())
            .hasSameElementsAs(storedInsertionIds.get(groupId));
    });
}
 

public GrayReleaseRulesHolder() {
  loadVersion = new AtomicLong();
  grayReleaseRuleCache = Multimaps.synchronizedSetMultimap(
      TreeMultimap.create(String.CASE_INSENSITIVE_ORDER, Ordering.natural()));
  reversedGrayReleaseRuleCache = Multimaps.synchronizedSetMultimap(
      TreeMultimap.create(String.CASE_INSENSITIVE_ORDER, Ordering.natural()));
  executorService = Executors.newScheduledThreadPool(1, ApolloThreadFactory
      .create("GrayReleaseRulesHolder", true));
}
 

private void init(ExecutorService executor) {
    tree = Maps.newTreeMap();
    if (executor != null) {
        this.executor = executor;
    } else {
        this.executor = Executors.newFixedThreadPool(1, new DefaultThreadFactory("mock-zookeeper"));
    }
    SetMultimap<String, Watcher> w = HashMultimap.create();
    watchers = Multimaps.synchronizedSetMultimap(w);
    stopped = false;
    alwaysFail = new AtomicReference<>(KeeperException.Code.OK);
    failures = new CopyOnWriteArrayList<>();
}
 

@Override
public void onSessionInitiated(ProviderContext session) {
	LOG.info("Router Session Initiated");
	routesWaitingOnArp = Multimaps.synchronizedSetMultimap(HashMultimap.<AtriumIpAddress, RouteEntry> create());
	ribTable4 = new ConcurrentInvertedRadixTree<>(new DefaultByteArrayNodeFactory());
	bgpUpdatesExecutor = Executors
			.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("atrium-bgp-updates-%d").build());
}
 

@Inject
public InVMEventBus(EventDelivery eventDelivery, RetryBackoffConfiguration retryBackoff, EventDeadLetters eventDeadLetters) {
    this.eventDelivery = eventDelivery;
    this.retryBackoff = retryBackoff;
    this.eventDeadLetters = eventDeadLetters;
    this.registrations = Multimaps.synchronizedSetMultimap(HashMultimap.create());
    this.groups = new ConcurrentHashMap<>();
}
 

public StateChangeNotifier(DAG dag) {
  this.dag = dag;
  this.vertexListeners = Multimaps.synchronizedSetMultimap(
      HashMultimap.<TezVertexID, ListenerContainer>create());
  this.lastKnowStatesMap = LinkedListMultimap.create();
  startThread();
}
 

private Multimap<String, ApplicationSLA> loadSchedulerMapCache(Map<String, ApplicationSLA> starterCache) {
    Multimap<String, ApplicationSLA> schedulerMapCache = Multimaps.synchronizedSetMultimap(HashMultimap.create());
    starterCache.values().forEach(applicationSLA -> schedulerMapCache.put(applicationSLA.getSchedulerName(), applicationSLA));
    return schedulerMapCache;
}
 

public MutableContextSetImpl() {
    this.map = Multimaps.synchronizedSetMultimap(HashMultimap.create());
}
 

MutableContextSetImpl(SetMultimap<String, String> other) {
    this.map = Multimaps.synchronizedSetMultimap(HashMultimap.create(other));
}
 

public InMemoryAttachmentMapper() {
    attachmentsById = new ConcurrentHashMap<>(INITIAL_SIZE);
    attachmentsRawContentById = new ConcurrentHashMap<>(INITIAL_SIZE);
    messageIdsByAttachmentId = Multimaps.synchronizedSetMultimap(HashMultimap.create());
    ownersByAttachmentId = Multimaps.synchronizedSetMultimap(HashMultimap.create());
}
 

public SegmentRouterInfo() {
    gatewayIps = Multimaps.synchronizedSetMultimap(HashMultimap.create());
    subnets = Multimaps.synchronizedSetMultimap(HashMultimap.create());
}
 

public CacheManager(int maximumSize) {
    backingCache = CacheBuilder.newBuilder().maximumSize(maximumSize)
                                            .removalListener(globalRemovalListener).build();
    segmentIndex = Multimaps.synchronizedSetMultimap(HashMultimap.<Object,CacheKey>create());
}
 

public static <T> MutableDirectedGraph<T> createConcurrent() {
  return new MutableDirectedGraph<>(
      Collections.newSetFromMap(new ConcurrentHashMap<T, Boolean>()),
      Multimaps.synchronizedSetMultimap(HashMultimap.create()),
      Multimaps.synchronizedSetMultimap(HashMultimap.create()));
}