import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
@Test
public void testSenderManager() {
    HostPort hostPort = new HostPort("192.168.1.10", 6379);
    Map<ALERT_TYPE, Set<AlertEntity>> alerts = new ConcurrentHashMap<>();
    AlertEntity alert = new AlertEntity(hostPort, dcNames[0], "cluster-test", "shard-test", "", ALERT_TYPE.XREDIS_VERSION_NOT_VALID);
    Set<AlertEntity> set = new ConcurrentSet<>();
    set.add(alert);
    alerts.put(ALERT_TYPE.XREDIS_VERSION_NOT_VALID, set);

    new Thread(new Runnable() {
        @Override
        public void run() {
            alerts.get(ALERT_TYPE.XREDIS_VERSION_NOT_VALID).remove(alert);
        }
    }).start();
    List<Map<ALERT_TYPE, Set<AlertEntity>>> result = senderManager.getGroupedAlerts(alerts);
    logger.info("result: {}", result.get(0));
    if(!result.isEmpty()) {
        Set<AlertEntity> alertEntities = result.get(0).getOrDefault(alert.getAlertType(), null);
        if(alertEntities != null) {
            Assert.assertFalse(alertEntities.isEmpty());
        }
    }
}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
@Test
public void listfield() throws CEUException, ParseException {

	ListHandler<String> handler = ListHandlerBuilder.build().byAllRows()
			.collect().byField("email", String.class);
	Command command = new Command("SELECT email FROM \"" + TABLE_NAME
			+ "\" allow filtering ");

	List<String> emails = existOptional(runner.query(command, handler));
	final Set<String> setEmails = new ConcurrentSet<String>();

	emails.stream().forEach(email -> {
		Assert.assertTrue(emails.get(0).startsWith("userTest"));
		Assert.assertTrue(emails.get(0).endsWith("@gmail.com"));
		setEmails.add(email);
	});

	Assert.assertEquals(emails.size() - 1, setEmails.size());

}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
public StreamCutService(final int bucketCount, String processId, final StreamMetadataStore streamMetadataStore,
                        final StreamMetadataTasks streamMetadataTasks, final ScheduledExecutorService executor) {
    this.bucketCount = bucketCount;
    this.processId = processId;
    this.streamMetadataStore = streamMetadataStore;
    this.streamMetadataTasks = streamMetadataTasks;
    this.executor = executor;
    this.buckets = new ConcurrentSet<>();
}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
@Override
public boolean add(Channel channel) {
    ConcurrentSet<Channel> set =
        channel instanceof ServerChannel? serverChannels : nonServerChannels;

    boolean added = set.add(channel);
    if (added) {
        channel.closeFuture().addListener(remover);
    }
    return added;
}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
/**
 * Initialize a new connection.
 *
 * @param writer the channel writer
 * @param codec Codec used to encode/decode keys and values.
 * @param timeout Maximum time to wait for a response.
 * @param unit Unit of time for the timeout.
 */
public StatefulRedisPubSubConnectionImpl(RedisChannelWriter<K, V> writer, RedisCodec<K, V> codec, long timeout,
        TimeUnit unit) {
    super(writer, codec, timeout, unit);

    listeners = new CopyOnWriteArrayList<>();
    channels = new ConcurrentSet<>();
    patterns = new ConcurrentSet<>();
}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
private Observable<Void> handleUpdateRecommendationsForUser(HttpServerRequest<ByteBuf> request, final HttpServerResponse<ByteBuf> response) {
    System.out.println("HTTP request -> update recommendations for user: " + request.getPath());
    final String userId = userIdFromPath(request.getPath());
    if (userId == null) {
        response.setStatus(HttpResponseStatus.BAD_REQUEST);
        return response.close();
    }
    return request.getContent().flatMap(new Func1<ByteBuf, Observable<Void>>() {
        @Override
        public Observable<Void> call(ByteBuf byteBuf) {
            String movieId = byteBuf.toString(Charset.defaultCharset());
            System.out.println(format("    updating: {user=%s, movie=%s}", userId, movieId));
            synchronized (this) {
                Set<String> recommendations;
                if (userRecommendations.containsKey(userId)) {
                    recommendations = userRecommendations.get(userId);
                } else {
                    recommendations = new ConcurrentSet<String>();
                    userRecommendations.put(userId, recommendations);
                }
                recommendations.add(movieId);
            }
            response.setStatus(HttpResponseStatus.OK);
            return response.close();
        }
    });
}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
@Override
public void onContextRefreshed() {
    runningJobs = new ConcurrentSet<>();
    executorService = Executors.newCachedThreadPool();
}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
@SuppressWarnings("MethodWithMultipleLoops")
@Test
public void deliver_updates_through_several_threads() throws InterruptedException {
    final int threadsCount = Given.THREADS_COUNT_IN_POOL_EXECUTOR;
    @SuppressWarnings("LocalVariableNamingConvention") // Too long variable name
    final int threadExecutionMaxAwaitSeconds = Given.AWAIT_SECONDS;

    final Set<String> threadInvocationRegistry = new ConcurrentSet<>();

    final Stand stand = Stand.newBuilder()
                             .build();

    final ExecutorService executor = Executors.newFixedThreadPool(threadsCount);

    final Runnable task = new Runnable() {
        @Override
        public void run() {
            final String threadName = Thread.currentThread()
                                            .getName();
            Assert.assertFalse(threadInvocationRegistry.contains(threadName));
            final ProjectId enitityId = ProjectId.newBuilder()
                                                 .setId(Identifier.newUuid())
                                                 .build();
            final Given.StandTestAggregate entity = Given.aggregateRepo()
                                                         .create(enitityId);
            stand.post(requestFactory.createCommandContext()
                                     .getActorContext()
                                     .getTenantId(), entity);

            threadInvocationRegistry.add(threadName);
        }
    };

    for (int i = 0; i < threadsCount; i++) {
        executor.execute(task);
    }

    executor.awaitTermination(threadExecutionMaxAwaitSeconds, TimeUnit.SECONDS);

    Assert.assertEquals(threadInvocationRegistry.size(), threadsCount);

}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
public HttpServerHandler(ConcurrentSet activeChannels, HttpServer server)
{
    this.server = server;
    this.activeChannels = activeChannels;
}
 

import io.netty.util.internal.ConcurrentSet; //导入依赖的package包/类
public HaProxyBackendServerHandler(ConcurrentSet activeChannels, HttpServer httpServer) {
    super(activeChannels, httpServer);
}