freemarker.template.Version#org.eclipse.jetty.util.BlockingArrayQueue源码实例Demo

@Override
public void init(Properties properties) {
  this.acceptors = Integer.parseInt(properties.getProperty(CONF_INFLUXDB_ACCEPTORS, "4"));
  this.selectors = Integer.parseInt(properties.getProperty(CONF_INFLUXDB_SELECTORS, "2"));
  this.maxThreads = Integer.parseInt(properties.getProperty(CONF_INFLUXDB_JETTY_THREADPOOL, Integer.toString(1 + acceptors + acceptors * selectors)));
  this.idleTimeout = Integer.parseInt(properties.getProperty(CONF_INFLUXDB_IDLE_TIMEOUT, "30000"));
  this.port = Integer.parseInt(properties.getProperty(CONF_INFLUXDB_PORT, "8086"));
  this.host = properties.getProperty(CONF_INFLUXDB_HOST, "127.0.0.1");
  this.token = properties.getProperty(CONF_INFLUXDB_DEFAULT_TOKEN);
  
  try {
    this.url = new URL(properties.getProperty(CONF_INFLUXDB_WARP10_ENDPOINT));
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
  
  if (properties.containsKey(CONF_INFLUXDB_JETTY_MAXQUEUESIZE)) {
    int queuesize = Integer.parseInt(properties.getProperty(CONF_INFLUXDB_JETTY_MAXQUEUESIZE));
    queue = new BlockingArrayQueue<Runnable>(queuesize);
  }
  
  Thread t = new Thread(this);
  t.setDaemon(true);
  t.setName("[InfluxDBWarp10Plugin " + host + ":" + port + "]");
  t.start();
}
 

@Override
protected void doStart() throws Exception
{
    super.doStart();
    _threadsStarted.set(0);

    if (_jobs==null)
    {
        _jobs=_maxQueued>0 ?new ArrayBlockingQueue<Runnable>(_maxQueued)
            :new BlockingArrayQueue<Runnable>(_minThreads,_minThreads);
    }

    int threads=_threadsStarted.get();
    while (isRunning() && threads<_minThreads)
    {
        startThread(threads);
        threads=_threadsStarted.get();
    }
}
 

@Override
protected void doStart() throws Exception
{
    super.doStart();
    _threadsStarted.set(0);

    if (_jobs==null)
    {
        _jobs=_maxQueued>0 ?new ArrayBlockingQueue<Runnable>(_maxQueued)
            :new BlockingArrayQueue<Runnable>(_minThreads,_minThreads);
    }

    int threads=_threadsStarted.get();
    while (isRunning() && threads<_minThreads)
    {
        startThread(threads);
        threads=_threadsStarted.get();
    }
}
 

/**
 * Create the thread pool with request queue.
 *
 * @return thread pool used by the server
 */
private static ThreadPool createThreadPool(RestConfig config) {
  /* Create blocking queue for the thread pool. */
  int initialCapacity = config.getInt(RestConfig.REQUEST_QUEUE_CAPACITY_INITIAL_CONFIG);
  int growBy = config.getInt(RestConfig.REQUEST_QUEUE_CAPACITY_GROWBY_CONFIG);
  int maxCapacity = config.getInt(RestConfig.REQUEST_QUEUE_CAPACITY_CONFIG);
  log.info("Initial capacity {}, increased by {}, maximum capacity {}.",
          initialCapacity, growBy, maxCapacity);

  BlockingQueue<Runnable> requestQueue =
          new BlockingArrayQueue<>(initialCapacity, growBy, maxCapacity);
  
  return new QueuedThreadPool(config.getInt(RestConfig.THREAD_POOL_MAX_CONFIG),
          config.getInt(RestConfig.THREAD_POOL_MIN_CONFIG),
          requestQueue);
}
 

@Override
protected void doStart() throws Exception
{
    super.doStart();
    _threadsStarted.set(0);

    if (_jobs==null)
    {
        _jobs=_maxQueued>0 ?new ArrayBlockingQueue<Runnable>(_maxQueued)
            :new BlockingArrayQueue<Runnable>(_minThreads,_minThreads);
    }

    int threads=_threadsStarted.get();
    while (isRunning() && threads<_minThreads)
    {
        startThread(threads);
        threads=_threadsStarted.get();
    }
}
 

@Test
public void updateUsingPostProcessConsumer() throws IOException {
    TestSetup ts = new TestSetup();

    BlockingQueue<Object> bc = new BlockingArrayQueue<>();
    PostProcessManager.Consumer consumer = new PostProcessManager.Consumer(bc, ts.getGraph(),
            getTypePropertyMap("hive_table", HIVE_COLUMNS_PROPERTY, "ARRAY"), 5);

    Vertex tableV = fetchTableVertex(ts.getGraph());
    consumer.processItem(tableV.id());
    ts.assertComplete();
}
 

private FreeMarkerService(Builder bulder) {
     maxOutputLength = bulder.getMaxOutputLength();
     maxThreads = bulder.getMaxThreads();
     maxQueueLength = bulder.getMaxQueueLength();
     maxTemplateExecutionTime = bulder.getMaxTemplateExecutionTime();

     int actualMaxQueueLength = maxQueueLength != null
             ? maxQueueLength
             : Math.max(
                     MIN_DEFAULT_MAX_QUEUE_LENGTH,
                     (int) (MAX_DEFAULT_MAX_QUEUE_LENGTH_MILLISECONDS / maxTemplateExecutionTime));
     ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
             maxThreads, maxThreads,
             THREAD_KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
             new BlockingArrayQueue<Runnable>(actualMaxQueueLength));
     threadPoolExecutor.allowCoreThreadTimeOut(true);
     templateExecutor = threadPoolExecutor;

     // Avoid ERROR log for using the actual current version. This application is special in that regard.
     Version latestVersion = new Version(Configuration.getVersion().toString());

     freeMarkerConfig = new Configuration(latestVersion);
     freeMarkerConfig.setNewBuiltinClassResolver(TemplateClassResolver.ALLOWS_NOTHING_RESOLVER);
     freeMarkerConfig.setObjectWrapper(new SimpleObjectWrapperWithXmlSupport(latestVersion));
     freeMarkerConfig.setTemplateExceptionHandler(TemplateExceptionHandler.RETHROW_HANDLER);
     freeMarkerConfig.setLogTemplateExceptions(false);
     freeMarkerConfig.setAttemptExceptionReporter(new AttemptExceptionReporter() {
@Override
public void report(TemplateException te, Environment env) {
	// Suppress it
}
     });
     freeMarkerConfig.setLocale(AllowedSettingValues.DEFAULT_LOCALE);
     freeMarkerConfig.setTimeZone(AllowedSettingValues.DEFAULT_TIME_ZONE);
     freeMarkerConfig.setOutputFormat(AllowedSettingValues.DEFAULT_OUTPUT_FORMAT);
     freeMarkerConfig.setOutputEncoding("UTF-8");
 }
 

@Override
public void init(Properties properties) {
  this.dir = properties.getProperty(CONF_HTTP_DIR);

  if (null == this.dir) {
    throw new RuntimeException("Missing '" + CONF_HTTP_DIR + "' configuration.");
  }

  this.period = Long.parseLong(properties.getProperty(CONF_HTTP_PERIOD, Long.toString(DEFAULT_PERIOD)));
  
  
  this.port = Integer.parseInt(properties.getProperty(CONF_HTTP_PORT, "-1"));
  this.tcpBacklog = Integer.parseInt(properties.getProperty(CONF_HTTP_TCP_BACKLOG, "0"));
  this.sslport = Integer.parseInt(properties.getProperty("http" + Configuration._SSL_PORT, "-1"));

  if (-1 == this.port && -1 == this.sslport) {
    throw new RuntimeException("Either '" + CONF_HTTP_PORT + "' or 'http." + Configuration._SSL_PORT + "' must be set.");
  }
  
  host = properties.getProperty(CONF_HTTP_HOST, null);
  acceptors = Integer.parseInt(properties.getProperty(CONF_HTTP_ACCEPTORS, String.valueOf(acceptors)));
  selectors = Integer.parseInt(properties.getProperty(CONF_HTTP_SELECTORS, String.valueOf(selectors)));
  idleTimeout = Integer.parseInt(properties.getProperty(CONF_HTTP_IDLE_TIMEOUT, String.valueOf(idleTimeout)));      

  maxthreads = Integer.parseInt(properties.getProperty(CONF_HTTP_MAXTHREADS, String.valueOf(maxthreads)));

  if (properties.containsKey(CONF_HTTP_QUEUESIZE)) {
    queue = new BlockingArrayQueue<Runnable>(Integer.parseInt(properties.getProperty(CONF_HTTP_QUEUESIZE)));
  }

  gzip = !"false".equals(properties.getProperty(CONF_HTTP_GZIP));
  lcheaders = "true".equals(properties.getProperty(CONF_HTTP_LCHEADERS));
  
  Thread t = new Thread(this);
  t.setDaemon(true);
  t.setName("[Warp 10 HTTP Plugin " + this.dir + "]");
  t.start();
}
 

/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
    afterPutEvts.clear();
    afterRmvEvts.clear();

    for (int i = 0; i < NODES; i++) {
        afterRmvEvts.put(grid(i).cluster().localNode().id(),
            new BlockingArrayQueue<Cache.Entry<TestKey, TestValue>>());
        afterPutEvts.put(grid(i).cluster().localNode().id(),
            new BlockingArrayQueue<Cache.Entry<TestKey, TestValue>>());
    }
}
 

@Test
public void testReplayOnConsumerDisconnect() throws Exception {
    final String topicName = "persistent://prop/use/ns-abc/shared-topic3";
    final String subName = "sub3";
    final int numMsgs = 100;

    final List<String> messagesProduced = Lists.newArrayListWithCapacity(numMsgs);
    final List<String> messagesConsumed = new BlockingArrayQueue<>(numMsgs);

    Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
            .subscriptionType(SubscriptionType.Shared).messageListener((consumer, msg) -> {
                try {
                    consumer.acknowledge(msg);
                    messagesConsumed.add(new String(msg.getData()));
                } catch (Exception e) {
                    fail("Should not fail");
                }
            }).subscribe();

    // consumer2 does not ack messages
    PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
    Consumer<byte[]> consumer2 = newPulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
            .subscriptionType(SubscriptionType.Shared).messageListener((consumer, msg) -> {
                // do notthing
            }).subscribe();

    List<CompletableFuture<MessageId>> futures = Lists.newArrayListWithCapacity(numMsgs * 2);
    Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
    for (int i = 0; i < numMsgs; i++) {
        String message = "msg-" + i;
        futures.add(producer.sendAsync(message.getBytes()));
        messagesProduced.add(message);
    }
    FutureUtil.waitForAll(futures).get();
    producer.close();

    consumer2.close();

    for (int n = 0; n < 10 && messagesConsumed.size() < numMsgs; n++) {
        Thread.sleep(ASYNC_EVENT_COMPLETION_WAIT);
    }

    // 1. consumer1 gets all messages
    assertTrue(CollectionUtils.subtract(messagesProduced, messagesConsumed).isEmpty());

    consumer1.close();
    newPulsarClient.close();

    deleteTopic(topicName);
}
 

/** {@inheritDoc} */
@Override public void onAfterPut(Cache.Entry<TestKey, TestValue> e) {
    e.getKey();
    e.getValue();

    UUID id = e.unwrap(Ignite.class).cluster().localNode().id();

    BlockingQueue<Cache.Entry<TestKey, TestValue>> ents = afterPutEvts.get(id);

    if (ents == null) {
        ents = new BlockingArrayQueue<>();

        BlockingQueue<Cache.Entry<TestKey, TestValue>> oldVal = afterPutEvts.putIfAbsent(id, ents);

        ents = oldVal == null ? ents : oldVal;
    }

    ents.add(e);
}
 

/** {@inheritDoc} */
@Override public void onAfterRemove(Cache.Entry<TestKey, TestValue> e) {
    e.getKey();
    e.getValue();

    UUID id = e.unwrap(Ignite.class).cluster().localNode().id();

    BlockingQueue<Cache.Entry<TestKey, TestValue>> ents = afterRmvEvts.get(id);

    if (ents == null) {
        ents = new BlockingArrayQueue<>();

        BlockingQueue<Cache.Entry<TestKey, TestValue>> oldVal = afterRmvEvts.putIfAbsent(id, ents);

        ents = oldVal == null ? ents : oldVal;
    }

    ents.add(e);
}
 

/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
    afterPutEvts = new BlockingArrayQueue<>();
    afterRmvEvts = new BlockingArrayQueue<>();
}
 

@Before
public void setUp() {
  BlockingQueue<Runnable> queue = new BlockingArrayQueue<>(8, 1024, 1024);
  queuedThreadPool = new QueuedThreadPool(200, 8, 60000, queue);
  server = new Server(queuedThreadPool);
}