com.google.common.collect.Queues#newLinkedBlockingDeque ( )源码实例Demo

PubSubClient(String hostname, int port, int maxPendingMessages) throws IOException {
    this.hostname = hostname;
    this.port = port;
    this.maxPendingMessages = maxPendingMessages;
    if (maxPendingMessages <= 0) {
        this.pending = Queues.newLinkedBlockingDeque();
    } else {
        this.pending = Queues.newLinkedBlockingDeque(maxPendingMessages);
    }
    this.selector = Selector.open();
    Runtime.getRuntime().addShutdownHook(new Thread() {
        public void run() {
            close();
        }
    });
}
 

private void registerBodyRequest(DtsResourceManager reousrceManager) {
    ResourceMessageHandler messageProcessor = new ResourceMessageHandler(reousrceManager);
    BlockingQueue<Runnable> clientThreadPoolQueue = Queues.newLinkedBlockingDeque(100);
    ExecutorService clientMessageExecutor =
        new ThreadPoolExecutor(nettyClientConfig.getClientCallbackExecutorThreads(),
            nettyClientConfig.getClientCallbackExecutorThreads(), 1000 * 60, TimeUnit.MILLISECONDS,
            clientThreadPoolQueue, new DtsThreadFactory("ResourceBodyRequestThread_"));
    this.remotingClient.registerProcessor(RequestCode.BODY_REQUEST, messageProcessor, clientMessageExecutor);
}
 

private void registerHeaderRequest(DtsResourceManager reousrceManager) {
    ResourceMessageHandler messageProcessor = new ResourceMessageHandler(reousrceManager);
    BlockingQueue<Runnable> clientThreadPoolQueue = Queues.newLinkedBlockingDeque(100);
    ExecutorService clientMessageExecutor =
        new ThreadPoolExecutor(nettyClientConfig.getClientCallbackExecutorThreads(),
            nettyClientConfig.getClientCallbackExecutorThreads(), 1000 * 60, TimeUnit.MILLISECONDS,
            clientThreadPoolQueue, new DtsThreadFactory("ResourceHeadRequestThread_"));
    this.remotingClient.registerProcessor(RequestCode.HEADER_REQUEST, messageProcessor, clientMessageExecutor);
}
 

private void registerBodyRequest() {
    DtsMessageProcessor messageProcessor = createMessageProcessor();
    BlockingQueue<Runnable> resourceThreadPoolQueue = Queues.newLinkedBlockingDeque(10000);
    ExecutorService resourceMessageExecutor = new ServerFixedThreadPoolExecutor(cpus, cpus, 1000 * 60,
        TimeUnit.MILLISECONDS, resourceThreadPoolQueue, new DtsThreadFactory("ServerBodyThread_"));
    this.remotingServer.registerProcessor(RequestCode.BODY_REQUEST, messageProcessor, resourceMessageExecutor);
}
 

private void registerHeaderRequest() {
    DtsMessageProcessor messageProcessor = createMessageProcessor();
    BlockingQueue<Runnable> clientThreadPoolQueue = Queues.newLinkedBlockingDeque(10000);
    ExecutorService clientMessageExecutor =
        new ServerFixedThreadPoolExecutor(cpus * headerRequestCorePoolSizeCpuTimes,
            cpus * headerRequestMaximumPoolSizeCpuTimes, headerRequestKeepaliveTime, TimeUnit.MILLISECONDS,
            clientThreadPoolQueue, new DtsThreadFactory("ServerHeadRequestThread_"));
    this.remotingServer.registerProcessor(RequestCode.HEADER_REQUEST, messageProcessor, clientMessageExecutor);
}
 

private Set<String> index() {
	Set<String> visited = Sets.newHashSet();
	Queue<Version> toVisit = Queues.newLinkedBlockingDeque();
	toVisit.add(rootVersion);

	while (!toVisit.isEmpty()) {
		Version current = toVisit.poll();
		if (visited.add(current.getId()) && current.getChild() != null) {
			toVisit.add(current.getChild());
		}
	}

	return visited;
}
 

/**
 * Constructor
 * 
 * @param tableName
 *            incoming table name
 * @param auths
 *            set of authorizations.
 * @param delegator
 *            scanner queue
 * @param maxResults
 */
public BatchScannerSession(String tableName, Set<Authorizations> auths, ResourceQueue delegator, int maxResults, Query settings, ScannerOptions options,
                Collection<Range> ranges) {
    
    super(tableName, auths, delegator, maxResults, settings);
    Preconditions.checkNotNull(delegator);
    
    localTableName = tableName;
    
    localAuths = auths;
    
    delegatorReference = super.sessionDelegator;
    
    scannerBatches = Iterators.emptyIterator();
    
    currentBatch = Queues.newLinkedBlockingDeque();
    
    setThreads(1);
    
    listenerService = Executors.newFixedThreadPool(1);
    
    addListener(new BatchScannerListener(), listenerService);
    
    serverFailureMap = Maps.newConcurrentMap();
    
    serverMap = Maps.newConcurrentMap();
    
}