From 678f30097b54c3703ad08d1133ada99baa9e0739 Mon Sep 17 00:00:00 2001 From: wu-sheng Date: Wed, 20 Dec 2017 14:30:15 +0800 Subject: [PATCH] Support to set consume cycle. And invoke the existed callbacks when the message blocked at first time. --- .../datacarrier/BlockingDataCarrier.java | 37 ++++++++++ .../apm/commons/datacarrier/DataCarrier.java | 67 ++++++++++++------- .../commons/datacarrier/buffer/Buffer.java | 16 ++++- .../commons/datacarrier/buffer/Channels.java | 15 +++-- .../callback/QueueBlockingCallback.java | 28 ++++++++ .../datacarrier/consumer/ConsumerPool.java | 13 ++-- .../datacarrier/consumer/ConsumerThread.java | 6 +- .../consumer/ConsumerPoolTest.java | 4 +- 8 files changed, 144 insertions(+), 42 deletions(-) create mode 100644 apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/BlockingDataCarrier.java create mode 100644 apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/callback/QueueBlockingCallback.java diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/BlockingDataCarrier.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/BlockingDataCarrier.java new file mode 100644 index 000000000..8b970808d --- /dev/null +++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/BlockingDataCarrier.java @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.commons.datacarrier; + +import org.apache.skywalking.apm.commons.datacarrier.buffer.Channels; +import org.apache.skywalking.apm.commons.datacarrier.callback.QueueBlockingCallback; + +/** + * @author wu-sheng + */ +public class BlockingDataCarrier { + private Channels channels; + + BlockingDataCarrier(Channels channels) { + this.channels = channels; + } + + public void addCallback(QueueBlockingCallback callback) { + this.channels.addCallback(callback); + } +} diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java index d40c1da82..081b25a6f 100644 --- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java +++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java @@ -16,21 +16,17 @@ * */ - package org.apache.skywalking.apm.commons.datacarrier; -import org.apache.skywalking.apm.commons.datacarrier.buffer.Channels; -import org.apache.skywalking.apm.commons.datacarrier.partition.IDataPartitioner; import org.apache.skywalking.apm.commons.datacarrier.buffer.BufferStrategy; +import org.apache.skywalking.apm.commons.datacarrier.buffer.Channels; import org.apache.skywalking.apm.commons.datacarrier.consumer.ConsumerPool; import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer; +import org.apache.skywalking.apm.commons.datacarrier.partition.IDataPartitioner; import org.apache.skywalking.apm.commons.datacarrier.partition.SimpleRollingPartitioner; /** - * DataCarrier main class. - * use this instance to set Producer/Consumer Model - *

- * Created by wusheng on 2016/10/25. + * DataCarrier main class. use this instance to set Producer/Consumer Model. */ public class DataCarrier { private final int bufferSize; @@ -45,8 +41,8 @@ public class DataCarrier { } /** - * set a new IDataPartitioner. - * It will cover the current one or default one.(Default is {@link SimpleRollingPartitioner)} + * set a new IDataPartitioner. It will cover the current one or default one.(Default is {@link + * SimpleRollingPartitioner)} * * @param dataPartitioner * @return @@ -57,8 +53,7 @@ public class DataCarrier { } /** - * override the strategy at runtime. - * Notice, {@link Channels} will override several channels one by one. + * override the strategy at runtime. Notice, {@link Channels} will override several channels one by one. * * @param strategy */ @@ -67,6 +62,11 @@ public class DataCarrier { return this; } + public BlockingDataCarrier toBlockingDataCarrier() { + this.channels.setStrategy(BufferStrategy.BLOCKING); + return new BlockingDataCarrier(this.channels); + } + /** * produce data to buffer, using the givven {@link BufferStrategy}. * @@ -84,36 +84,57 @@ public class DataCarrier { } /** - * set consumers to this Carrier. - * consumer begin to run when {@link DataCarrier#produce(T)} begin to work. + * set consumers to this Carrier. consumer begin to run when {@link DataCarrier#produce(T)} begin to work. * * @param consumerClass class of consumer * @param num number of consumer threads */ - public DataCarrier consume(Class> consumerClass, int num) { + public DataCarrier consume(Class> consumerClass, int num, long consumeCycle) { if (consumerPool != null) { consumerPool.close(); } - consumerPool = new ConsumerPool(this.channels, consumerClass, num); + consumerPool = new ConsumerPool(this.channels, consumerClass, num, consumeCycle); consumerPool.begin(); return this; } /** - * set consumers to this Carrier. - * consumer begin to run when {@link DataCarrier#produce(T)} begin to work. + * set consumers to this Carrier. consumer begin to run when {@link DataCarrier#produce(T)} begin to work with 20 + * millis consume cycle. + * + * @param consumerClass class of consumer + * @param num number of consumer threads + */ + public DataCarrier consume(Class> consumerClass, int num) { + return this.consume(consumerClass, num, 20); + } + + /** + * set consumers to this Carrier. consumer begin to run when {@link DataCarrier#produce(T)} begin to work. + * + * @param consumer single instance of consumer, all consumer threads will all use this instance. + * @param num number of consumer threads + * @return + */ + public DataCarrier consume(IConsumer consumer, int num, long consumeCycle) { + if (consumerPool != null) { + consumerPool.close(); + } + consumerPool = new ConsumerPool(this.channels, consumer, num, consumeCycle); + consumerPool.begin(); + return this; + } + + /** + * set consumers to this Carrier. consumer begin to run when {@link DataCarrier#produce(T)} begin to work with 20 + * millis consume cycle. * * @param consumer single instance of consumer, all consumer threads will all use this instance. * @param num number of consumer threads * @return */ public DataCarrier consume(IConsumer consumer, int num) { - if (consumerPool != null) { - consumerPool.close(); - } - consumerPool = new ConsumerPool(this.channels, consumer, num); - consumerPool.begin(); - return this; + return this.consume(consumer, num, 20); } /** diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Buffer.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Buffer.java index 6ac79c479..4fbb47869 100644 --- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Buffer.java +++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Buffer.java @@ -16,10 +16,11 @@ * */ - package org.apache.skywalking.apm.commons.datacarrier.buffer; import java.util.LinkedList; +import java.util.List; +import org.apache.skywalking.apm.commons.datacarrier.callback.QueueBlockingCallback; import org.apache.skywalking.apm.commons.datacarrier.common.AtomicRangeInteger; /** @@ -29,23 +30,36 @@ public class Buffer { private final Object[] buffer; private BufferStrategy strategy; private AtomicRangeInteger index; + private List> callbacks; Buffer(int bufferSize, BufferStrategy strategy) { buffer = new Object[bufferSize]; this.strategy = strategy; index = new AtomicRangeInteger(0, bufferSize); + callbacks = new LinkedList>(); } void setStrategy(BufferStrategy strategy) { this.strategy = strategy; } + void addCallback(QueueBlockingCallback callback) { + callbacks.add(callback); + } + boolean save(T data) { int i = index.getAndIncrement(); if (buffer[i] != null) { switch (strategy) { case BLOCKING: + boolean isFirstTimeBlocking = true; while (buffer[i] != null) { + if (isFirstTimeBlocking) { + isFirstTimeBlocking = false; + for (QueueBlockingCallback callback : callbacks) { + callback.notify(data); + } + } try { Thread.sleep(1L); } catch (InterruptedException e) { diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Channels.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Channels.java index bf4c03fd7..4f7ee8754 100644 --- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Channels.java +++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Channels.java @@ -16,17 +16,14 @@ * */ - package org.apache.skywalking.apm.commons.datacarrier.buffer; +import org.apache.skywalking.apm.commons.datacarrier.callback.QueueBlockingCallback; import org.apache.skywalking.apm.commons.datacarrier.partition.IDataPartitioner; /** - * Channels of Buffer - * It contais all buffer data which belongs to this channel. - * It supports several strategy when buffer is full. The Default is BLOCKING - *

- * Created by wusheng on 2016/10/25. + * Channels of Buffer It contais all buffer data which belongs to this channel. It supports several strategy when buffer + * is full. The Default is BLOCKING

Created by wusheng on 2016/10/25. */ public class Channels { private final Buffer[] bufferChannels; @@ -87,4 +84,10 @@ public class Channels { public Buffer getBuffer(int index) { return this.bufferChannels[index]; } + + public void addCallback(QueueBlockingCallback callback) { + for (Buffer channel : bufferChannels) { + channel.addCallback(callback); + } + } } diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/callback/QueueBlockingCallback.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/callback/QueueBlockingCallback.java new file mode 100644 index 000000000..c0a6c47c4 --- /dev/null +++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/callback/QueueBlockingCallback.java @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.commons.datacarrier.callback; + +/** + * Notify when the queue, which is in blocking strategy, has be blocked. + * + * @author wu-sheng + */ +public interface QueueBlockingCallback { + void notify(T message); +} diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java index 6d127cca7..814fdf8f8 100644 --- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java +++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java @@ -16,7 +16,6 @@ * */ - package org.apache.skywalking.apm.commons.datacarrier.consumer; import java.util.ArrayList; @@ -25,9 +24,7 @@ import org.apache.skywalking.apm.commons.datacarrier.buffer.Buffer; import org.apache.skywalking.apm.commons.datacarrier.buffer.Channels; /** - * Pool of consumers - *

- * Created by wusheng on 2016/10/25. + * Pool of consumers

Created by wusheng on 2016/10/25. */ public class ConsumerPool { private boolean running; @@ -35,19 +32,19 @@ public class ConsumerPool { private Channels channels; private ReentrantLock lock; - public ConsumerPool(Channels channels, Class> consumerClass, int num) { + public ConsumerPool(Channels channels, Class> consumerClass, int num, long consumeCycle) { this(channels, num); for (int i = 0; i < num; i++) { - consumerThreads[i] = new ConsumerThread("DataCarrier.Consumser." + i + ".Thread", getNewConsumerInstance(consumerClass)); + consumerThreads[i] = new ConsumerThread("DataCarrier.Consumser." + i + ".Thread", getNewConsumerInstance(consumerClass), consumeCycle); consumerThreads[i].setDaemon(true); } } - public ConsumerPool(Channels channels, IConsumer prototype, int num) { + public ConsumerPool(Channels channels, IConsumer prototype, int num, long consumeCycle) { this(channels, num); prototype.init(); for (int i = 0; i < num; i++) { - consumerThreads[i] = new ConsumerThread("DataCarrier.Consumser." + i + ".Thread", prototype); + consumerThreads[i] = new ConsumerThread("DataCarrier.Consumser." + i + ".Thread", prototype, consumeCycle); consumerThreads[i].setDaemon(true); } diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java index 891aac465..1ba0223db 100644 --- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java +++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java @@ -30,12 +30,14 @@ public class ConsumerThread extends Thread { private volatile boolean running; private IConsumer consumer; private List dataSources; + private long consumeCycle; - ConsumerThread(String threadName, IConsumer consumer) { + ConsumerThread(String threadName, IConsumer consumer, long consumeCycle) { super(threadName); this.consumer = consumer; running = false; dataSources = new LinkedList(); + this.consumeCycle = consumeCycle; } /** @@ -67,7 +69,7 @@ public class ConsumerThread extends Thread { if (!hasData) { try { - Thread.sleep(20); + Thread.sleep(consumeCycle); } catch (InterruptedException e) { } } diff --git a/apm-commons/apm-datacarrier/src/test/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPoolTest.java b/apm-commons/apm-datacarrier/src/test/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPoolTest.java index 4daa51e4f..18eab0eac 100644 --- a/apm-commons/apm-datacarrier/src/test/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPoolTest.java +++ b/apm-commons/apm-datacarrier/src/test/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPoolTest.java @@ -34,7 +34,7 @@ public class ConsumerPoolTest { @Test public void testBeginConsumerPool() throws IllegalAccessException { Channels channels = new Channels(2, 100, new SimpleRollingPartitioner(), BufferStrategy.BLOCKING); - ConsumerPool pool = new ConsumerPool(channels, new SampleConsumer(), 2); + ConsumerPool pool = new ConsumerPool(channels, new SampleConsumer(), 2, 20); pool.begin(); ConsumerThread[] threads = (ConsumerThread[])MemberModifier.field(ConsumerPool.class, "consumerThreads").get(pool); @@ -46,7 +46,7 @@ public class ConsumerPoolTest { @Test public void testCloseConsumerPool() throws InterruptedException, IllegalAccessException { Channels channels = new Channels(2, 100, new SimpleRollingPartitioner(), BufferStrategy.BLOCKING); - ConsumerPool pool = new ConsumerPool(channels, new SampleConsumer(), 2); + ConsumerPool pool = new ConsumerPool(channels, new SampleConsumer(), 2, 20); pool.begin(); Thread.sleep(5000);