metricsDataToWarehouseRedisQueue;
+
+ public MetricsDataExporter() {
+ metricsDataToAlertQueue = new LinkedBlockingQueue<>();
+ metricsDataToWarehouseInfluxQueue = new LinkedBlockingQueue<>();
+ metricsDataToWarehouseRedisQueue = new LinkedBlockingQueue<>();
+ }
+
+ public CollectRep.MetricsData pollAlertMetricsData() throws InterruptedException {
+ return metricsDataToAlertQueue.poll(2, TimeUnit.SECONDS);
+ }
+
+ public CollectRep.MetricsData pollWarehouseInfluxMetricsData() throws InterruptedException {
+ return metricsDataToAlertQueue.poll(2, TimeUnit.SECONDS);
+ }
+
+ public CollectRep.MetricsData pollWarehouseRedisMetricsData() throws InterruptedException {
+ return metricsDataToWarehouseRedisQueue.poll(2, TimeUnit.SECONDS);
+ }
+
+ /**
+ * 发送消息
+ * @param metricsData 指标组采集数据
+ */
+ public void send(CollectRep.MetricsData metricsData) {
+ metricsDataToAlertQueue.offer(metricsData);
+ metricsDataToWarehouseInfluxQueue.offer(metricsData);
+ metricsDataToWarehouseRedisQueue.offer(metricsData);
+ }
+
+ @Override
+ public void destroy() throws Exception {
+ metricsDataToAlertQueue.clear();
+ }
+}
diff --git a/collector/src/main/java/com/usthe/collector/dispatch/timer/HashedWheelTimer.java b/collector/src/main/java/com/usthe/collector/dispatch/timer/HashedWheelTimer.java
new file mode 100644
index 0000000..9e71563
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/dispatch/timer/HashedWheelTimer.java
@@ -0,0 +1,809 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 com.usthe.collector.dispatch.timer;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * A {@link Timer} optimized for approximated I/O timeout scheduling.
+ *
+ * Tick Duration
+ *
+ * As described with 'approximated', this timer does not execute the scheduled
+ * {@link TimerTask} on time. {@link HashedWheelTimer}, on every tick, will
+ * check if there are any {@link TimerTask}s behind the schedule and execute
+ * them.
+ *
+ * You can increase or decrease the accuracy of the execution timing by
+ * specifying smaller or larger tick duration in the constructor. In most
+ * network applications, I/O timeout does not need to be accurate. Therefore,
+ * the default tick duration is 100 milliseconds and you will not need to try
+ * different configurations in most cases.
+ *
+ *
Ticks per Wheel (Wheel Size)
+ *
+ * {@link HashedWheelTimer} maintains a data structure called 'wheel'.
+ * To put simply, a wheel is a hash table of {@link TimerTask}s whose hash
+ * function is 'dead line of the task'. The default number of ticks per wheel
+ * (i.e. the size of the wheel) is 512. You could specify a larger value
+ * if you are going to schedule a lot of timeouts.
+ *
+ *
Do not create many instances.
+ *
+ * {@link HashedWheelTimer} creates a new thread whenever it is instantiated and
+ * started. Therefore, you should make sure to create only one instance and
+ * share it across your application. One of the common mistakes, that makes
+ * your application unresponsive, is to create a new instance for every connection.
+ *
+ *
Implementation Details
+ *
+ * {@link HashedWheelTimer} is based on
+ * George Varghese and
+ * Tony Lauck's paper,
+ * 'Hashed
+ * and Hierarchical Timing Wheels: data structures to efficiently implement a
+ * timer facility'. More comprehensive slides are located
+ * here.
+ * @author from netty | dubbo
+ */
+@SuppressWarnings("PMD")
+public class HashedWheelTimer implements Timer {
+
+ private static final Logger logger = LoggerFactory.getLogger(HashedWheelTimer.class);
+
+ private static final AtomicInteger INSTANCE_COUNTER = new AtomicInteger();
+ private static final AtomicBoolean WARNED_TOO_MANY_INSTANCES = new AtomicBoolean();
+ private static final int INSTANCE_COUNT_LIMIT = 64;
+ private static final AtomicIntegerFieldUpdater WORKER_STATE_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");
+
+ private final Worker worker = new Worker();
+ private final Thread workerThread;
+
+ private static final int WORKER_STATE_INIT = 0;
+ private static final int WORKER_STATE_STARTED = 1;
+ private static final int WORKER_STATE_SHUTDOWN = 2;
+
+ /**
+ * 0 - init, 1 - started, 2 - shut down
+ */
+ @SuppressWarnings({"unused", "FieldMayBeFinal"})
+ private volatile int workerState;
+
+ private final long tickDuration;
+ private final HashedWheelBucket[] wheel;
+ private final int mask;
+ private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
+ private final Queue timeouts = new LinkedBlockingQueue<>();
+ private final Queue cancelledTimeouts = new LinkedBlockingQueue<>();
+ private final AtomicLong pendingTimeouts = new AtomicLong(0);
+ private final long maxPendingTimeouts;
+
+ private volatile long startTime;
+
+ /**
+ * Creates a new timer with the default thread factory
+ * ({@link Executors#defaultThreadFactory()}), default tick duration, and
+ * default number of ticks per wheel.
+ */
+ public HashedWheelTimer() {
+ this(Executors.defaultThreadFactory());
+ }
+
+ /**
+ * Creates a new timer with the default thread factory
+ * ({@link Executors#defaultThreadFactory()}) and default number of ticks
+ * per wheel.
+ *
+ * @param tickDuration the duration between tick
+ * @param unit the time unit of the {@code tickDuration}
+ * @throws NullPointerException if {@code unit} is {@code null}
+ * @throws IllegalArgumentException if {@code tickDuration} is <= 0
+ */
+ public HashedWheelTimer(long tickDuration, TimeUnit unit) {
+ this(Executors.defaultThreadFactory(), tickDuration, unit);
+ }
+
+ /**
+ * Creates a new timer with the default thread factory
+ * ({@link Executors#defaultThreadFactory()}).
+ *
+ * @param tickDuration the duration between tick
+ * @param unit the time unit of the {@code tickDuration}
+ * @param ticksPerWheel the size of the wheel
+ * @throws NullPointerException if {@code unit} is {@code null}
+ * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
+ */
+ public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
+ this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
+ }
+
+ /**
+ * Creates a new timer with the default tick duration and default number of
+ * ticks per wheel.
+ *
+ * @param threadFactory a {@link ThreadFactory} that creates a
+ * background {@link Thread} which is dedicated to
+ * {@link TimerTask} execution.
+ * @throws NullPointerException if {@code threadFactory} is {@code null}
+ */
+ public HashedWheelTimer(ThreadFactory threadFactory) {
+ this(threadFactory, 100, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Creates a new timer with the default number of ticks per wheel.
+ *
+ * @param threadFactory a {@link ThreadFactory} that creates a
+ * background {@link Thread} which is dedicated to
+ * {@link TimerTask} execution.
+ * @param tickDuration the duration between tick
+ * @param unit the time unit of the {@code tickDuration}
+ * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}
+ * @throws IllegalArgumentException if {@code tickDuration} is <= 0
+ */
+ public HashedWheelTimer(
+ ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
+ this(threadFactory, tickDuration, unit, 512);
+ }
+
+ /**
+ * Creates a new timer.
+ *
+ * @param threadFactory a {@link ThreadFactory} that creates a
+ * background {@link Thread} which is dedicated to
+ * {@link TimerTask} execution.
+ * @param tickDuration the duration between tick
+ * @param unit the time unit of the {@code tickDuration}
+ * @param ticksPerWheel the size of the wheel
+ * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}
+ * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
+ */
+ public HashedWheelTimer(
+ ThreadFactory threadFactory,
+ long tickDuration, TimeUnit unit, int ticksPerWheel) {
+ this(threadFactory, tickDuration, unit, ticksPerWheel, -1);
+ }
+
+ /**
+ * Creates a new timer.
+ *
+ * @param threadFactory a {@link ThreadFactory} that creates a
+ * background {@link Thread} which is dedicated to
+ * {@link TimerTask} execution.
+ * @param tickDuration the duration between tick
+ * @param unit the time unit of the {@code tickDuration}
+ * @param ticksPerWheel the size of the wheel
+ * @param maxPendingTimeouts The maximum number of pending timeouts after which call to
+ * {@code newTimeout} will result in
+ * {@link RejectedExecutionException}
+ * being thrown. No maximum pending timeouts limit is assumed if
+ * this value is 0 or negative.
+ * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}
+ * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
+ */
+ public HashedWheelTimer(
+ ThreadFactory threadFactory,
+ long tickDuration, TimeUnit unit, int ticksPerWheel,
+ long maxPendingTimeouts) {
+
+ if (threadFactory == null) {
+ throw new NullPointerException("threadFactory");
+ }
+ if (unit == null) {
+ throw new NullPointerException("unit");
+ }
+ if (tickDuration <= 0) {
+ throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
+ }
+ if (ticksPerWheel <= 0) {
+ throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
+ }
+
+ // Normalize ticksPerWheel to power of two and initialize the wheel.
+ wheel = createWheel(ticksPerWheel);
+ mask = wheel.length - 1;
+
+ // Convert tickDuration to nanos.
+ this.tickDuration = unit.toNanos(tickDuration);
+
+ // Prevent overflow.
+ if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {
+ throw new IllegalArgumentException(String.format(
+ "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
+ tickDuration, Long.MAX_VALUE / wheel.length));
+ }
+ workerThread = threadFactory.newThread(worker);
+
+ this.maxPendingTimeouts = maxPendingTimeouts;
+
+ if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
+ WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
+ reportTooManyInstances();
+ }
+ }
+
+ @Override
+ protected void finalize() throws Throwable {
+ try {
+ super.finalize();
+ } finally {
+ // This object is going to be GCed and it is assumed the ship has sailed to do a proper shutdown. If
+ // we have not yet shutdown then we want to make sure we decrement the active instance count.
+ if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
+ INSTANCE_COUNTER.decrementAndGet();
+ }
+ }
+ }
+
+ private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
+ if (ticksPerWheel <= 0) {
+ throw new IllegalArgumentException(
+ "ticksPerWheel must be greater than 0: " + ticksPerWheel);
+ }
+ if (ticksPerWheel > 1073741824) {
+ throw new IllegalArgumentException(
+ "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);
+ }
+
+ ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
+ HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
+ for (int i = 0; i < wheel.length; i++) {
+ wheel[i] = new HashedWheelBucket();
+ }
+ return wheel;
+ }
+
+ private static int normalizeTicksPerWheel(int ticksPerWheel) {
+ int normalizedTicksPerWheel = ticksPerWheel - 1;
+ normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 1;
+ normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 2;
+ normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 4;
+ normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 8;
+ normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 16;
+ return normalizedTicksPerWheel + 1;
+ }
+
+ /**
+ * Starts the background thread explicitly. The background thread will
+ * start automatically on demand even if you did not call this method.
+ *
+ * @throws IllegalStateException if this timer has been
+ * {@linkplain #stop() stopped} already
+ */
+ public void start() {
+ switch (WORKER_STATE_UPDATER.get(this)) {
+ case WORKER_STATE_INIT:
+ if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
+ workerThread.start();
+ }
+ break;
+ case WORKER_STATE_STARTED:
+ break;
+ case WORKER_STATE_SHUTDOWN:
+ throw new IllegalStateException("cannot be started once stopped");
+ default:
+ throw new Error("Invalid WorkerState");
+ }
+
+ // Wait until the startTime is initialized by the worker.
+ while (startTime == 0) {
+ try {
+ startTimeInitialized.await();
+ } catch (InterruptedException ignore) {
+ // Ignore - it will be ready very soon.
+ }
+ }
+ }
+
+ @Override
+ public Set stop() {
+ if (Thread.currentThread() == workerThread) {
+ throw new IllegalStateException(
+ HashedWheelTimer.class.getSimpleName() +
+ ".stop() cannot be called from " +
+ TimerTask.class.getSimpleName());
+ }
+
+ if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
+ // workerState can be 0 or 2 at this moment - let it always be 2.
+ if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
+ INSTANCE_COUNTER.decrementAndGet();
+ }
+
+ return Collections.emptySet();
+ }
+
+ try {
+ boolean interrupted = false;
+ while (workerThread.isAlive()) {
+ workerThread.interrupt();
+ try {
+ workerThread.join(100);
+ } catch (InterruptedException ignored) {
+ interrupted = true;
+ }
+ }
+
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ } finally {
+ INSTANCE_COUNTER.decrementAndGet();
+ }
+ return worker.unprocessedTimeouts();
+ }
+
+ @Override
+ public boolean isStop() {
+ return WORKER_STATE_SHUTDOWN == WORKER_STATE_UPDATER.get(this);
+ }
+
+ @Override
+ public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
+ if (task == null) {
+ throw new NullPointerException("task");
+ }
+ if (unit == null) {
+ throw new NullPointerException("unit");
+ }
+
+ long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
+
+ if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
+ pendingTimeouts.decrementAndGet();
+ throw new RejectedExecutionException("Number of pending timeouts ("
+ + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
+ + "timeouts (" + maxPendingTimeouts + ")");
+ }
+
+ start();
+
+ // Add the timeout to the timeout queue which will be processed on the next tick.
+ // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
+ long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
+
+ // Guard against overflow.
+ if (delay > 0 && deadline < 0) {
+ deadline = Long.MAX_VALUE;
+ }
+ HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
+ timeouts.add(timeout);
+ return timeout;
+ }
+
+ /**
+ * Returns the number of pending timeouts of this {@link Timer}.
+ */
+ public long pendingTimeouts() {
+ return pendingTimeouts.get();
+ }
+
+ private static void reportTooManyInstances() {
+ logger.error("You are creating too many HashedWheelTimer instances. " +
+ "HashedWheelTimer is a shared resource that must be reused across the JVM," +
+ "so that only a few instances are created.");
+ }
+
+ private final class Worker implements Runnable {
+ private final Set unprocessedTimeouts = new HashSet();
+
+ private long tick;
+
+ @Override
+ public void run() {
+ // Initialize the startTime.
+ startTime = System.nanoTime();
+ if (startTime == 0) {
+ // We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
+ startTime = 1;
+ }
+
+ // Notify the other threads waiting for the initialization at start().
+ startTimeInitialized.countDown();
+
+ do {
+ final long deadline = waitForNextTick();
+ if (deadline > 0) {
+ int idx = (int) (tick & mask);
+ processCancelledTasks();
+ HashedWheelBucket bucket =
+ wheel[idx];
+ transferTimeoutsToBuckets();
+ bucket.expireTimeouts(deadline);
+ tick++;
+ }
+ } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
+
+ // Fill the unprocessedTimeouts so we can return them from stop() method.
+ for (HashedWheelBucket bucket : wheel) {
+ bucket.clearTimeouts(unprocessedTimeouts);
+ }
+ for (; ; ) {
+ HashedWheelTimeout timeout = timeouts.poll();
+ if (timeout == null) {
+ break;
+ }
+ if (!timeout.isCancelled()) {
+ unprocessedTimeouts.add(timeout);
+ }
+ }
+ processCancelledTasks();
+ }
+
+ private void transferTimeoutsToBuckets() {
+ // transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just
+ // adds new timeouts in a loop.
+ for (int i = 0; i < 100000; i++) {
+ HashedWheelTimeout timeout = timeouts.poll();
+ if (timeout == null) {
+ // all processed
+ break;
+ }
+ if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
+ // Was cancelled in the meantime.
+ continue;
+ }
+
+ long calculated = timeout.deadline / tickDuration;
+ timeout.remainingRounds = (calculated - tick) / wheel.length;
+
+ // Ensure we don't schedule for past.
+ final long ticks = Math.max(calculated, tick);
+ int stopIndex = (int) (ticks & mask);
+
+ HashedWheelBucket bucket = wheel[stopIndex];
+ bucket.addTimeout(timeout);
+ }
+ }
+
+ private void processCancelledTasks() {
+ for (; ; ) {
+ HashedWheelTimeout timeout = cancelledTimeouts.poll();
+ if (timeout == null) {
+ // all processed
+ break;
+ }
+ try {
+ timeout.remove();
+ } catch (Throwable t) {
+ if (logger.isWarnEnabled()) {
+ logger.warn("An exception was thrown while process a cancellation task", t);
+ }
+ }
+ }
+ }
+
+ /**
+ * calculate goal nanoTime from startTime and current tick number,
+ * then wait until that goal has been reached.
+ *
+ * @return Long.MIN_VALUE if received a shutdown request,
+ * current time otherwise (with Long.MIN_VALUE changed by +1)
+ */
+ private long waitForNextTick() {
+ long deadline = tickDuration * (tick + 1);
+
+ for (; ; ) {
+ final long currentTime = System.nanoTime() - startTime;
+ long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
+
+ if (sleepTimeMs <= 0) {
+ if (currentTime == Long.MIN_VALUE) {
+ return -Long.MAX_VALUE;
+ } else {
+ return currentTime;
+ }
+ }
+ if (isWindows()) {
+ sleepTimeMs = sleepTimeMs / 10 * 10;
+ }
+
+ try {
+ Thread.sleep(sleepTimeMs);
+ } catch (InterruptedException ignored) {
+ if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
+ return Long.MIN_VALUE;
+ }
+ }
+ }
+ }
+
+ Set unprocessedTimeouts() {
+ return Collections.unmodifiableSet(unprocessedTimeouts);
+ }
+ }
+
+ private static final class HashedWheelTimeout implements Timeout {
+
+ private static final int ST_INIT = 0;
+ private static final int ST_CANCELLED = 1;
+ private static final int ST_EXPIRED = 2;
+ private static final AtomicIntegerFieldUpdater STATE_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
+
+ private final HashedWheelTimer timer;
+ private final TimerTask task;
+ private final long deadline;
+
+ @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization"})
+ private volatile int state = ST_INIT;
+
+ /**
+ * RemainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the
+ * HashedWheelTimeout will be added to the correct HashedWheelBucket.
+ */
+ long remainingRounds;
+
+ /**
+ * This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list.
+ * As only the workerThread will act on it there is no need for synchronization / volatile.
+ */
+ HashedWheelTimeout next;
+ HashedWheelTimeout prev;
+
+ /**
+ * The bucket to which the timeout was added
+ */
+ HashedWheelBucket bucket;
+
+ HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
+ this.timer = timer;
+ this.task = task;
+ this.deadline = deadline;
+ }
+
+ @Override
+ public Timer timer() {
+ return timer;
+ }
+
+ @Override
+ public TimerTask task() {
+ return task;
+ }
+
+ @Override
+ public boolean cancel() {
+ // only update the state it will be removed from HashedWheelBucket on next tick.
+ if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
+ return false;
+ }
+ // If a task should be canceled we put this to another queue which will be processed on each tick.
+ // So this means that we will have a GC latency of max. 1 tick duration which is good enough. This way
+ // we can make again use of our MpscLinkedQueue and so minimize the locking / overhead as much as possible.
+ timer.cancelledTimeouts.add(this);
+ return true;
+ }
+
+ void remove() {
+ HashedWheelBucket bucket = this.bucket;
+ if (bucket != null) {
+ bucket.remove(this);
+ } else {
+ timer.pendingTimeouts.decrementAndGet();
+ }
+ }
+
+ public boolean compareAndSetState(int expected, int state) {
+ return STATE_UPDATER.compareAndSet(this, expected, state);
+ }
+
+ public int state() {
+ return state;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return state() == ST_CANCELLED;
+ }
+
+ @Override
+ public boolean isExpired() {
+ return state() == ST_EXPIRED;
+ }
+
+ public void expire() {
+ if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
+ return;
+ }
+
+ try {
+ task.run(this);
+ } catch (Throwable t) {
+ if (logger.isWarnEnabled()) {
+ logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ final long currentTime = System.nanoTime();
+ long remaining = deadline - currentTime + timer.startTime;
+
+ StringBuilder buf = new StringBuilder(192)
+ .append("HashedWheelTimer")
+ .append('(')
+ .append("deadline: ");
+ if (remaining > 0) {
+ buf.append(remaining)
+ .append(" ns later");
+ } else if (remaining < 0) {
+ buf.append(-remaining)
+ .append(" ns ago");
+ } else {
+ buf.append("now");
+ }
+
+ if (isCancelled()) {
+ buf.append(", cancelled");
+ }
+
+ return buf.append(", task: ")
+ .append(task())
+ .append(')')
+ .toString();
+ }
+ }
+
+ /**
+ * Bucket that stores HashedWheelTimeouts. These are stored in a linked-list like datastructure to allow easy
+ * removal of HashedWheelTimeouts in the middle. Also the HashedWheelTimeout act as nodes themself and so no
+ * extra object creation is needed.
+ */
+ private static final class HashedWheelBucket {
+
+ /**
+ * Used for the linked-list datastructure
+ */
+ private HashedWheelTimeout head;
+ private HashedWheelTimeout tail;
+
+ /**
+ * Add {@link HashedWheelTimeout} to this bucket.
+ */
+ void addTimeout(HashedWheelTimeout timeout) {
+ assert timeout.bucket == null;
+ timeout.bucket = this;
+ if (head == null) {
+ head = tail = timeout;
+ } else {
+ tail.next = timeout;
+ timeout.prev = tail;
+ tail = timeout;
+ }
+ }
+
+ /**
+ * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
+ */
+ void expireTimeouts(long deadline) {
+ HashedWheelTimeout timeout = head;
+
+ // process all timeouts
+ while (timeout != null) {
+ HashedWheelTimeout next = timeout.next;
+ if (timeout.remainingRounds <= 0) {
+ next = remove(timeout);
+ if (timeout.deadline <= deadline) {
+ timeout.expire();
+ } else {
+ // The timeout was placed into a wrong slot. This should never happen.
+ throw new IllegalStateException(String.format(
+ "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
+ }
+ } else if (timeout.isCancelled()) {
+ next = remove(timeout);
+ } else {
+ timeout.remainingRounds--;
+ }
+ timeout = next;
+ }
+ }
+
+ public HashedWheelTimeout remove(HashedWheelTimeout timeout) {
+ HashedWheelTimeout next = timeout.next;
+ // remove timeout that was either processed or cancelled by updating the linked-list
+ if (timeout.prev != null) {
+ timeout.prev.next = next;
+ }
+ if (timeout.next != null) {
+ timeout.next.prev = timeout.prev;
+ }
+
+ if (timeout == head) {
+ // if timeout is also the tail we need to adjust the entry too
+ if (timeout == tail) {
+ tail = null;
+ head = null;
+ } else {
+ head = next;
+ }
+ } else if (timeout == tail) {
+ // if the timeout is the tail modify the tail to be the prev node.
+ tail = timeout.prev;
+ }
+ // null out prev, next and bucket to allow for GC.
+ timeout.prev = null;
+ timeout.next = null;
+ timeout.bucket = null;
+ timeout.timer.pendingTimeouts.decrementAndGet();
+ return next;
+ }
+
+ /**
+ * Clear this bucket and return all not expired / cancelled {@link Timeout}s.
+ */
+ void clearTimeouts(Set set) {
+ for (; ; ) {
+ HashedWheelTimeout timeout = pollTimeout();
+ if (timeout == null) {
+ return;
+ }
+ if (timeout.isExpired() || timeout.isCancelled()) {
+ continue;
+ }
+ set.add(timeout);
+ }
+ }
+
+ private HashedWheelTimeout pollTimeout() {
+ HashedWheelTimeout head = this.head;
+ if (head == null) {
+ return null;
+ }
+ HashedWheelTimeout next = head.next;
+ if (next == null) {
+ tail = this.head = null;
+ } else {
+ this.head = next;
+ next.prev = null;
+ }
+
+ // null out prev and next to allow for GC.
+ head.next = null;
+ head.prev = null;
+ head.bucket = null;
+ return head;
+ }
+ }
+
+ private static final boolean IS_OS_WINDOWS = System.getProperty("os.name", "").toLowerCase(Locale.US).contains("win");
+
+ private boolean isWindows() {
+ return IS_OS_WINDOWS;
+ }
+}
diff --git a/collector/src/main/java/com/usthe/collector/dispatch/timer/Timeout.java b/collector/src/main/java/com/usthe/collector/dispatch/timer/Timeout.java
new file mode 100644
index 0000000..f82a6a3
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/dispatch/timer/Timeout.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 com.usthe.collector.dispatch.timer;
+
+/**
+ * A handle associated with a {@link TimerTask} that is returned by a
+ * {@link Timer}.
+ * @author from netty | dubbo
+ */
+@SuppressWarnings("PMD")
+public interface Timeout {
+
+ /**
+ * Returns the {@link Timer} that created this handle.
+ */
+ Timer timer();
+
+ /**
+ * Returns the {@link TimerTask} which is associated with this handle.
+ */
+ TimerTask task();
+
+ /**
+ * Returns {@code true} if and only if the {@link TimerTask} associated
+ * with this handle has been expired.
+ */
+ boolean isExpired();
+
+ /**
+ * Returns {@code true} if and only if the {@link TimerTask} associated
+ * with this handle has been cancelled.
+ */
+ boolean isCancelled();
+
+ /**
+ * Attempts to cancel the {@link TimerTask} associated with this handle.
+ * If the task has been executed or cancelled already, it will return with
+ * no side effect.
+ *
+ * @return True if the cancellation completed successfully, otherwise false
+ */
+ boolean cancel();
+}
diff --git a/collector/src/main/java/com/usthe/collector/dispatch/timer/Timer.java b/collector/src/main/java/com/usthe/collector/dispatch/timer/Timer.java
new file mode 100644
index 0000000..e962752
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/dispatch/timer/Timer.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 com.usthe.collector.dispatch.timer;
+
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Schedules {@link TimerTask}s for one-time future execution in a background
+ * thread.
+ * @author from netty | dubbo
+ */
+public interface Timer {
+
+ /**
+ * Schedules the specified {@link TimerTask} for one-time execution after
+ * the specified delay.
+ *
+ * @param task the {@link TimerTask
+ * @param delay the delay
+ * @param unit the unit of time
+ * @return a handle which is associated with the specified task
+ * @throws IllegalStateException if this timer has been {@linkplain #stop() stopped} already
+ * @throws RejectedExecutionException if the pending timeouts are too many and creating new timeout
+ * can cause instability in the system.
+ */
+ Timeout newTimeout(TimerTask task, long delay, TimeUnit unit);
+
+ /**
+ * Releases all resources acquired by this {@link Timer} and cancels all
+ * tasks which were scheduled but not executed yet.
+ *
+ * @return the handles associated with the tasks which were canceled by
+ * this method
+ */
+ Set stop();
+
+ /**
+ * the timer is stop
+ *
+ * @return true for stop
+ */
+ boolean isStop();
+}
diff --git a/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerDispatch.java b/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerDispatch.java
new file mode 100644
index 0000000..578e504
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerDispatch.java
@@ -0,0 +1,46 @@
+package com.usthe.collector.dispatch.timer;
+
+
+import com.usthe.collector.dispatch.entrance.internal.CollectResponseEventListener;
+import com.usthe.common.entity.job.Job;
+import com.usthe.common.entity.message.CollectRep;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 时间轮调度接口
+ * @author tomsun28
+ * @date 2021/10/17 22:14
+ */
+public interface TimerDispatch {
+
+ /**
+ * 增加新的job
+ * @param addJob job
+ * @param eventListener 一次性同步任务监听器，异步任务不需要listener
+ */
+ void addJob(Job addJob, CollectResponseEventListener eventListener);
+
+ /**
+ * 调度循环周期性job
+ * @param timerTask timerTask
+ * @param interval 开始调度的间隔时间
+ * @param timeUnit 时间单位
+ */
+ void cyclicJob(WheelTimerTask timerTask, long interval, TimeUnit timeUnit);
+
+ /**
+ * 删除存在的job
+ * @param jobId jobId
+ * @param isCyclic 是否是周期性任务,true是, false为临时性任务
+ */
+ void deleteJob(long jobId, boolean isCyclic);
+
+ /**
+ * 一次性同步采集任务采集结果通知监听器
+ * @param jobId jobId
+ * @param metricsDataTemps 采集结果数据
+ */
+ void responseSyncJobData(long jobId, List metricsDataTemps);
+}
diff --git a/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerDispatcher.java b/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerDispatcher.java
new file mode 100644
index 0000000..46a4b05
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerDispatcher.java
@@ -0,0 +1,95 @@
+package com.usthe.collector.dispatch.timer;
+
+import com.usthe.collector.dispatch.entrance.internal.CollectResponseEventListener;
+import com.usthe.common.entity.job.Job;
+import com.usthe.common.entity.message.CollectRep;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author tomsun28
+ * @date 2021/10/17 23:06
+ */
+@Component
+public class TimerDispatcher implements TimerDispatch {
+
+ /**
+ * 时间轮调度
+ */
+ private Timer wheelTimer;
+ /**
+ * 已存在的周期性调度任务
+ */
+ private Map currentCyclicTaskMap;
+ /**
+ * 已存在的临时性调度任务
+ */
+ private Map currentTempTaskMap;
+ /**
+ * 一次性任务响应监听器持有
+ * jobId - listener
+ */
+ private Map eventListeners;
+
+ public TimerDispatcher() {
+ this.wheelTimer = new HashedWheelTimer(r -> {
+ Thread ret = new Thread(r, "wheelTimer");
+ ret.setDaemon(true);
+ return ret;
+ }, 10, TimeUnit.SECONDS, 512);
+ this.currentCyclicTaskMap = new ConcurrentHashMap<>(1024);
+ this.currentTempTaskMap = new ConcurrentHashMap<>(64);
+ eventListeners = new ConcurrentHashMap<>(64);
+ }
+
+ @Override
+ public void addJob(Job addJob, CollectResponseEventListener eventListener) {
+ WheelTimerTask timerJob = new WheelTimerTask(addJob);
+ if (addJob.isCyclic()) {
+ Timeout timeout = wheelTimer.newTimeout(timerJob, addJob.getInterval(), TimeUnit.SECONDS);
+ currentCyclicTaskMap.put(addJob.getId(), timeout);
+ } else {
+ Timeout timeout = wheelTimer.newTimeout(timerJob, 0, TimeUnit.SECONDS);
+ currentTempTaskMap.put(addJob.getId(), timeout);
+ eventListeners.put(addJob.getId(), eventListener);
+ }
+ }
+
+ @Override
+ public void cyclicJob(WheelTimerTask timerTask, long interval, TimeUnit timeUnit) {
+ Long jobId = timerTask.getJob().getId();
+ // 判断此周期性job是否已经被取消
+ if (currentCyclicTaskMap.containsKey(jobId)) {
+ Timeout timeout = wheelTimer.newTimeout(timerTask, interval, TimeUnit.SECONDS);
+ currentCyclicTaskMap.put(timerTask.getJob().getId(), timeout);
+ }
+ }
+
+ @Override
+ public void deleteJob(long jobId, boolean isCyclic) {
+ if (isCyclic) {
+ Timeout timeout = currentCyclicTaskMap.remove(jobId);
+ if (timeout != null) {
+ timeout.cancel();
+ }
+ } else {
+ Timeout timeout = currentTempTaskMap.remove(jobId);
+ if (timeout != null) {
+ timeout.cancel();
+ }
+ }
+ }
+
+ @Override
+ public void responseSyncJobData(long jobId, List metricsDataTemps) {
+ currentTempTaskMap.remove(jobId);
+ CollectResponseEventListener eventListener = eventListeners.remove(jobId);
+ if (eventListener != null) {
+ eventListener.response(metricsDataTemps);
+ }
+ }
+}
diff --git a/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerTask.java b/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerTask.java
new file mode 100644
index 0000000..bec8423
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/dispatch/timer/TimerTask.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project 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 com.usthe.collector.dispatch.timer;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A task which is executed after the delay specified with
+ * {@link Timer#newTimeout(TimerTask, long, TimeUnit)} (TimerTask, long, TimeUnit)}.
+ * @author from netty | dubbo
+ */
+public interface TimerTask {
+
+ /**
+ * Executed after the delay specified with
+ * {@link Timer#newTimeout(TimerTask, long, TimeUnit)}.
+ *
+ * @param timeout a handle which is associated with this task
+ * @throws Exception when error happen
+ */
+ void run(Timeout timeout) throws Exception;
+}
diff --git a/collector/src/main/java/com/usthe/collector/dispatch/timer/WheelTimerTask.java b/collector/src/main/java/com/usthe/collector/dispatch/timer/WheelTimerTask.java
new file mode 100644
index 0000000..6998ffe
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/dispatch/timer/WheelTimerTask.java
@@ -0,0 +1,125 @@
+package com.usthe.collector.dispatch.timer;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
+import com.usthe.collector.dispatch.MetricsTaskDispatch;
+import com.usthe.collector.util.SpringContextHolder;
+import com.usthe.common.entity.job.Configmap;
+import com.usthe.common.entity.job.Job;
+import com.usthe.common.entity.job.Metrics;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * TimerTask实现
+ * @author tomsun28
+ * @date 2021/11/1 17:18
+ */
+public class WheelTimerTask implements TimerTask {
+
+ private final Job job;
+ private final MetricsTaskDispatch metricsTaskDispatch;
+ private static final Gson GSON = new Gson();
+
+ public WheelTimerTask(Job job) {
+ this.metricsTaskDispatch = SpringContextHolder.getBean(MetricsTaskDispatch.class);
+ this.job = job;
+ // 初始化job 将监控实际参数值对采集字段进行替换
+ initJobMetrics(job);
+ }
+
+ /**
+ * 初始化job填充信息
+ * @param job job
+ */
+ private void initJobMetrics(Job job) {
+ // 将监控实际参数值对采集字段进行替换
+ List config = job.getConfigmap();
+ Map configmap = config.stream().collect(Collectors.toMap(Configmap::getKey, item -> item));
+ List metrics = job.getMetrics();
+ List metricsTmp = new ArrayList<>(metrics.size());
+ for (Metrics metric : metrics) {
+ JsonElement jsonElement = GSON.toJsonTree(metric);
+ jsonElement = replaceSpecialValue(jsonElement, configmap);
+ metric = GSON.fromJson(jsonElement, Metrics.class);
+ metricsTmp.add(metric);
+ }
+ job.setMetrics(metricsTmp);
+ }
+
+ /**
+ * json参数替换
+ * @param jsonElement json
+ * @param configmap 参数map
+ * @return json
+ */
+ private JsonElement replaceSpecialValue(JsonElement jsonElement, Map configmap) {
+ if (jsonElement.isJsonObject()) {
+ JsonObject jsonObject = jsonElement.getAsJsonObject();
+ Iterator> iterator = jsonObject.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry entry = iterator.next();
+ JsonElement element = entry.getValue();
+ if (element.isJsonPrimitive()) {
+ // 判断是否含有特殊字符 替换
+ String value = element.getAsString();
+ if (value.startsWith("^_^") && value.endsWith("^_^")) {
+ value = value.replaceAll("\\^_\\^", "");
+ Configmap param = configmap.get(value);
+ if (param != null) {
+ value = (String) param.getValue();
+ jsonObject.addProperty(entry.getKey(), value);
+ } else {
+ iterator.remove();
+ }
+ }
+ } else {
+ jsonObject.add(entry.getKey(), replaceSpecialValue(entry.getValue(), configmap));
+ }
+ }
+ } else if (jsonElement.isJsonArray()) {
+ JsonArray jsonArray = jsonElement.getAsJsonArray();
+ Iterator iterator = jsonArray.iterator();
+ int index = 0;
+ while (iterator.hasNext()) {
+ JsonElement element = iterator.next();
+ if (element.isJsonPrimitive()) {
+ // 判断是否含有特殊字符 替换
+ String value = element.getAsString();
+ if (value.startsWith("^_^") && value.endsWith("^_^")) {
+ value = value.replaceAll("\\^_\\^", "");
+ Configmap param = configmap.get(value);
+ if (param != null) {
+ value = (String) param.getValue();
+ jsonArray.set(index, new JsonPrimitive(value));
+ } else {
+ iterator.remove();
+ }
+ }
+ } else {
+ jsonArray.set(index, replaceSpecialValue(element, configmap));
+ }
+ index++;
+ }
+ }
+ return jsonElement;
+ }
+
+
+ @Override
+ public void run(Timeout timeout) throws Exception {
+ job.setDispatchTime(System.currentTimeMillis());
+ metricsTaskDispatch.dispatchMetricsTask(timeout);
+ }
+
+ public Job getJob() {
+ return job;
+ }
+}
diff --git a/collector/src/main/java/com/usthe/collector/util/CollectorConstants.java b/collector/src/main/java/com/usthe/collector/util/CollectorConstants.java
new file mode 100644
index 0000000..176acd3
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/util/CollectorConstants.java
@@ -0,0 +1,11 @@
+package com.usthe.collector.util;
+
+/**
+ * collector 常量
+ * @author tom
+ * @date 2021/12/3 12:15
+ */
+public interface CollectorConstants {
+
+ String RESPONSE_TIME = "responseTime";
+}
diff --git a/collector/src/main/java/com/usthe/collector/util/JsonPathParser.java b/collector/src/main/java/com/usthe/collector/util/JsonPathParser.java
new file mode 100644
index 0000000..88e78b7
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/util/JsonPathParser.java
@@ -0,0 +1,44 @@
+package com.usthe.collector.util;
+
+import com.jayway.jsonpath.Configuration;
+import com.jayway.jsonpath.JsonPath;
+import com.jayway.jsonpath.Option;
+import com.jayway.jsonpath.ParseContext;
+import com.jayway.jsonpath.spi.cache.CacheProvider;
+import com.jayway.jsonpath.spi.cache.LRUCache;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * json path parser
+ * @author tomsun28
+ * @date 2021/11/20 10:16
+ */
+public class JsonPathParser {
+
+ private static final ParseContext PARSER;
+
+ static {
+ Configuration conf = Configuration.defaultConfiguration()
+ .addOptions(Option.DEFAULT_PATH_LEAF_TO_NULL)
+ .addOptions(Option.ALWAYS_RETURN_LIST);
+ CacheProvider.setCache(new LRUCache(128));
+ PARSER = JsonPath.using(conf);
+ }
+
+ /**
+ * 使用jsonPath来解析json内容
+ * @param content json内容
+ * @param jsonPath jsonPath脚本
+ * @return 解析后的内容
+ */
+ public static List> parseContentWithJsonPath(String content, String jsonPath) {
+ if (content == null || jsonPath == null || "".equals(content) || "".equals(jsonPath)) {
+ return Collections.emptyList();
+ }
+ return PARSER.parse(content).read(jsonPath);
+ }
+
+}
diff --git a/collector/src/main/java/com/usthe/collector/util/SpringContextHolder.java b/collector/src/main/java/com/usthe/collector/util/SpringContextHolder.java
new file mode 100644
index 0000000..ef3a85f
--- /dev/null
+++ b/collector/src/main/java/com/usthe/collector/util/SpringContextHolder.java
@@ -0,0 +1,48 @@
+package com.usthe.collector.util;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+/**
+ * Spring的ApplicationContext的持有者,可以用静态方法的方式获取spring容器中的bean
+ * @author tomsun28
+ * @date 21:07 2018/4/18
+ */
+@Component
+public class SpringContextHolder implements ApplicationContextAware {
+
+ private static ApplicationContext applicationContext;
+
+ @Override
+ public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+ set(applicationContext);
+ }
+
+ private static void set(ApplicationContext applicationContext) {
+ SpringContextHolder.applicationContext = applicationContext;
+ }
+
+ public static ApplicationContext getApplicationContext() {
+ assertApplicationContext();
+ return applicationContext;
+ }
+
+ @SuppressWarnings("unchecked")
+ public static T getBean(String beanName) {
+ assertApplicationContext();
+ return (T) applicationContext.getBean(beanName);
+ }
+
+ public static T getBean(Class tClass) {
+ assertApplicationContext();
+ return (T) applicationContext.getBean(tClass);
+ }
+
+ private static void assertApplicationContext() {
+ if (null == SpringContextHolder.applicationContext) {
+ throw new RuntimeException("applicationContext为空,请检查是否注入springContextHolder");
+ }
+ }
+}
diff --git a/collector/src/main/resources/META-INF/spring.factories b/collector/src/main/resources/META-INF/spring.factories
new file mode 100644
index 0000000..6f385a1
--- /dev/null
+++ b/collector/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,9 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+com.usthe.collector.dispatch.timer.TimerDispatcher,\
+com.usthe.collector.dispatch.CommonDispatcher,\
+com.usthe.collector.dispatch.DispatchProperties,\
+com.usthe.collector.dispatch.MetricsCollectorQueue,\
+com.usthe.collector.dispatch.WorkerPool,\
+com.usthe.collector.dispatch.entrance.internal.CollectJobService,\
+com.usthe.collector.dispatch.export.MetricsDataExporter,\
+com.usthe.collector.util.SpringContextHolder
diff --git a/collector/src/test/java/com/usthe/collector/collect/telnet/TelnetCollectImplTest.java b/collector/src/test/java/com/usthe/collector/collect/telnet/TelnetCollectImplTest.java
new file mode 100644
index 0000000..4c4e92a
--- /dev/null
+++ b/collector/src/test/java/com/usthe/collector/collect/telnet/TelnetCollectImplTest.java
@@ -0,0 +1,37 @@
+package com.usthe.collector.collect.telnet;
+
+import org.apache.commons.net.telnet.TelnetClient;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * @author tom
+ * @date 2021/12/4 19:39
+ */
+class TelnetCollectImplTest {
+
+ @Test
+ void telnet() {
+ TelnetClient telnetClient = null;
+ try {
+ telnetClient = new TelnetClient("vt200");
+ telnetClient.setConnectTimeout(5000);
+ TelnetClient finalTelnetClient = telnetClient;
+ assertDoesNotThrow(() -> finalTelnetClient.connect("baidu.com",80));
+ telnetClient.disconnect();
+ } catch (IOException e) {
+ e.printStackTrace();
+ } finally {
+ if (telnetClient != null) {
+ try {
+ telnetClient.disconnect();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/common/pom.xml b/common/pom.xml
index 65949be..5679795 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -5,7 +5,7 @@
monitor
com.usthe.tancloud
- 1.0-SNAPSHOT
+ 1.0
4.0.0
diff --git a/manager/pom.xml b/manager/pom.xml
index d6da03c..ecca547 100644
--- a/manager/pom.xml
+++ b/manager/pom.xml
@@ -5,7 +5,7 @@
monitor
com.usthe.tancloud
- 1.0-SNAPSHOT
+ 1.0
4.0.0
manager
@@ -22,25 +22,25 @@
com.usthe.tancloud
common
- 1.0-SNAPSHOT
-
-
-
- com.usthe.tancloud
- scheduler
- 1.0-SNAPSHOT
+ 1.0
com.usthe.tancloud
warehouse
- 1.0-SNAPSHOT
+ 1.0
com.usthe.tancloud
alerter
- 1.0-SNAPSHOT
+ 1.0
+
+
+
+ com.usthe.tancloud
+ collector
+ 1.0
@@ -123,6 +123,7 @@
sureness.yml
banner.txt
db/**
+ define/**
@@ -167,7 +168,7 @@
- ../assembly/server/assembly.xml
+ ../script/assembly/server/assembly.xml
diff --git a/manager/src/main/java/com/usthe/manager/service/impl/MonitorServiceImpl.java b/manager/src/main/java/com/usthe/manager/service/impl/MonitorServiceImpl.java
index 024d003..2a6e9ca 100644
--- a/manager/src/main/java/com/usthe/manager/service/impl/MonitorServiceImpl.java
+++ b/manager/src/main/java/com/usthe/manager/service/impl/MonitorServiceImpl.java
@@ -1,5 +1,6 @@
package com.usthe.manager.service.impl;
+import com.usthe.collector.dispatch.entrance.internal.CollectJobService;
import com.usthe.common.entity.job.Configmap;
import com.usthe.common.entity.job.Job;
import com.usthe.common.entity.job.Metrics;
@@ -20,7 +21,6 @@ import com.usthe.manager.service.AppService;
import com.usthe.manager.service.MonitorService;
import com.usthe.manager.support.exception.MonitorDatabaseException;
import com.usthe.manager.support.exception.MonitorDetectException;
-import com.usthe.scheduler.JobScheduling;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
@@ -52,7 +52,7 @@ public class MonitorServiceImpl implements MonitorService {
private AppService appService;
@Autowired
- private JobScheduling jobScheduling;
+ private CollectJobService collectJobService;
@Autowired
private MonitorDao monitorDao;
@@ -74,7 +74,7 @@ public class MonitorServiceImpl implements MonitorService {
List configmaps = params.stream().map(param ->
new Configmap(param.getField(), param.getValue(), param.getType())).collect(Collectors.toList());
appDefine.setConfigmap(configmaps);
- List collectRep = jobScheduling.addSyncCollectJob(appDefine);
+ List collectRep = collectJobService.collectSyncJobData(appDefine);
// 判断探测结果 失败则抛出探测异常
if (collectRep == null || collectRep.isEmpty()) {
throw new MonitorDetectException("No collector response");
@@ -101,7 +101,7 @@ public class MonitorServiceImpl implements MonitorService {
}).collect(Collectors.toList());
appDefine.setConfigmap(configmaps);
// 下发采集任务得到jobId
- long jobId = jobScheduling.addAsyncCollectJob(appDefine);
+ long jobId = collectJobService.addAsyncCollectJob(appDefine);
// 下发成功后刷库
try {
monitor.setId(monitorId);
@@ -112,7 +112,7 @@ public class MonitorServiceImpl implements MonitorService {
} catch (Exception e) {
log.error(e.getMessage(), e);
// 刷库异常取消之前的下发任务
- jobScheduling.cancelAsyncCollectJob(jobId);
+ collectJobService.cancelAsyncCollectJob(jobId);
throw new MonitorDatabaseException(e.getMessage());
}
}
@@ -225,7 +225,7 @@ public class MonitorServiceImpl implements MonitorService {
new Configmap(param.getField(), param.getValue(), param.getType())).collect(Collectors.toList());
appDefine.setConfigmap(configmaps);
// 更新采集任务
- jobScheduling.updateAsyncCollectJob(appDefine);
+ collectJobService.updateAsyncCollectJob(appDefine);
// 下发更新成功后刷库
try {
monitor.setJobId(preMonitor.getJobId());
@@ -246,7 +246,7 @@ public class MonitorServiceImpl implements MonitorService {
Monitor monitor = monitorOptional.get();
monitorDao.deleteById(id);
paramDao.deleteParamsByMonitorId(id);
- jobScheduling.cancelAsyncCollectJob(monitor.getJobId());
+ collectJobService.cancelAsyncCollectJob(monitor.getJobId());
}
}
@@ -258,7 +258,7 @@ public class MonitorServiceImpl implements MonitorService {
monitorDao.deleteAll(monitors);
paramDao.deleteParamsByMonitorIdIn(ids);
for (Monitor monitor : monitors) {
- jobScheduling.cancelAsyncCollectJob(monitor.getJobId());
+ collectJobService.cancelAsyncCollectJob(monitor.getJobId());
}
}
}
@@ -299,7 +299,7 @@ public class MonitorServiceImpl implements MonitorService {
if (!managedMonitors.isEmpty()) {
monitorDao.saveAll(managedMonitors);
for (Monitor monitor : managedMonitors) {
- jobScheduling.cancelAsyncCollectJob(monitor.getJobId());
+ collectJobService.cancelAsyncCollectJob(monitor.getJobId());
}
}
}
@@ -326,7 +326,7 @@ public class MonitorServiceImpl implements MonitorService {
new Configmap(param.getField(), param.getValue(), param.getType())).collect(Collectors.toList());
appDefine.setConfigmap(configmaps);
// 下发采集任务
- jobScheduling.addAsyncCollectJob(appDefine);
+ collectJobService.addAsyncCollectJob(appDefine);
}
}
}
diff --git a/manager/src/main/java/com/usthe/manager/support/GlobalExceptionHandler.java b/manager/src/main/java/com/usthe/manager/support/GlobalExceptionHandler.java
index 271234c..cde3378 100644
--- a/manager/src/main/java/com/usthe/manager/support/GlobalExceptionHandler.java
+++ b/manager/src/main/java/com/usthe/manager/support/GlobalExceptionHandler.java
@@ -4,7 +4,6 @@ package com.usthe.manager.support;
import com.usthe.common.entity.dto.Message;
import com.usthe.manager.support.exception.MonitorDatabaseException;
import com.usthe.manager.support.exception.MonitorDetectException;
-import com.usthe.scheduler.ScheduleException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataAccessException;
import org.springframework.http.HttpStatus;
@@ -133,23 +132,6 @@ public class GlobalExceptionHandler {
return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(message);
}
- /**
- * 处理分发调度器异常问题
- * @param exception 调度器异常问题
- * @return response
- */
- @ExceptionHandler(ScheduleException.class)
- @ResponseBody
- ResponseEntity> handleScheduleException(ScheduleException exception) {
- String errorMessage = "scheduler warning";
- if (exception != null) {
- errorMessage = exception.getMessage();
- }
- log.warn("[scheduler warning]-{}", errorMessage);
- Message message = Message.builder().msg(errorMessage).code(MONITOR_CONFLICT_CODE).build();
- return ResponseEntity.status(HttpStatus.CONFLICT).body(message);
- }
-
/**
* handler the exception thrown for datastore error
* @param exception datastore exception
diff --git a/pom.xml b/pom.xml
index ab68842..c4da00a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -7,9 +7,8 @@
com.usthe.tancloud
monitor
pom
- 1.0-SNAPSHOT
+ 1.0
- scheduler
manager
alerter
common
diff --git a/script/assembly/package-build.sh b/script/assembly/package-build.sh
new file mode 100644
index 0000000..4bd63c6
--- /dev/null
+++ b/script/assembly/package-build.sh
@@ -0,0 +1,9 @@
+#!/bin/bash
+
+cd ../../web-app
+
+ng build --prod --base-href /console/
+
+cd ..
+
+mvn clean package
diff --git a/assembly/server/assembly.xml b/script/assembly/server/assembly.xml
similarity index 91%
rename from assembly/server/assembly.xml
rename to script/assembly/server/assembly.xml
index 038731d..91d5399 100644
--- a/assembly/server/assembly.xml
+++ b/script/assembly/server/assembly.xml
@@ -22,8 +22,11 @@ http://maven.apache.org/ASSEMBLY/2.0.0 ">
- ../assembly/server/bin
+ ../script/assembly/server/bin
+
+ true
bin
+ 0755
diff --git a/assembly/collector/bin/shutdown.sh b/script/assembly/server/bin/shutdown.sh
similarity index 100%
rename from assembly/collector/bin/shutdown.sh
rename to script/assembly/server/bin/shutdown.sh
diff --git a/assembly/server/bin/startup.sh b/script/assembly/server/bin/startup.sh
similarity index 100%
rename from assembly/server/bin/startup.sh
rename to script/assembly/server/bin/startup.sh
diff --git a/warehouse/pom.xml b/warehouse/pom.xml
index eb4f04d..25bc65b 100644
--- a/warehouse/pom.xml
+++ b/warehouse/pom.xml
@@ -5,7 +5,7 @@
monitor
com.usthe.tancloud
- 1.0-SNAPSHOT
+ 1.0
4.0.0
@@ -16,7 +16,14 @@
com.usthe.tancloud
common
- 1.0-SNAPSHOT
+ 1.0
+
+
+
+ com.usthe.tancloud
+ collector
+ 1.0
+ provided
diff --git a/warehouse/src/main/java/com/usthe/warehouse/entrance/KafkaDataConsume.java b/warehouse/src/main/java/com/usthe/warehouse/entrance/KafkaDataConsume.java
deleted file mode 100644
index ad99bed..0000000
--- a/warehouse/src/main/java/com/usthe/warehouse/entrance/KafkaDataConsume.java
+++ /dev/null
@@ -1,81 +0,0 @@
-package com.usthe.warehouse.entrance;
-
-import com.usthe.common.entity.message.CollectRep;
-import com.usthe.warehouse.MetricsDataQueue;
-import com.usthe.warehouse.WarehouseProperties;
-import com.usthe.warehouse.WarehouseWorkerPool;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.serialization.LongDeserializer;
-import org.springframework.beans.factory.DisposableBean;
-import org.springframework.boot.autoconfigure.AutoConfigureAfter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Configuration;
-
-import java.time.Duration;
-import java.util.Collections;
-import java.util.Properties;
-
-/**
- * 从Kafka消费指标组采集数据处理
- * @author tom
- * @date 2021/11/24 18:03
- */
-@Configuration
-@AutoConfigureAfter(value = {WarehouseProperties.class})
-@ConditionalOnProperty(prefix = "warehouse.entrance.kafka",
- name = "enabled", havingValue = "true", matchIfMissing = true)
-@Slf4j
-public class KafkaDataConsume implements DisposableBean {
-
- private KafkaConsumer consumer;
- private WarehouseWorkerPool workerPool;
- private MetricsDataQueue dataQueue;
- public KafkaDataConsume(WarehouseProperties properties, WarehouseWorkerPool workerPool,
- MetricsDataQueue dataQueue) {
- this.workerPool = workerPool;
- this.dataQueue = dataQueue;
- initConsumer(properties);
- startConsumeData();
- }
-
- private void startConsumeData() {
- Runnable runnable = () -> {
- Thread.currentThread().setName("warehouse-kafka-data-consumer");
- while (!Thread.currentThread().isInterrupted()) {
- ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
- records.forEach(record -> {
- dataQueue.addMetricsDataToInflux(record.value());
- dataQueue.addMetricsDataToRedis(record.value());
- });
- }
- };
- workerPool.executeJob(runnable);
- }
-
- private void initConsumer(WarehouseProperties properties) {
- if (properties == null || properties.getEntrance() == null || properties.getEntrance().getKafka() == null) {
- log.error("init error, please config Warehouse kafka props in application.yml");
- throw new IllegalArgumentException("please config Warehouse kafka props");
- }
- WarehouseProperties.EntranceProperties.KafkaProperties kafkaProp = properties.getEntrance().getKafka();
- Properties consumerProp = new Properties();
- consumerProp.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProp.getServers());
- consumerProp.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProp.getGroupId());
- consumerProp.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
- consumerProp.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaMetricsDataDeserializer.class);
- consumerProp.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
- consumerProp.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
- consumer = new KafkaConsumer<>(consumerProp);
- consumer.subscribe(Collections.singleton(kafkaProp.getTopic()));
- }
-
- @Override
- public void destroy() throws Exception {
- if (consumer != null) {
- consumer.close();
- }
- }
-}
diff --git a/warehouse/src/main/java/com/usthe/warehouse/entrance/KafkaMetricsDataDeserializer.java b/warehouse/src/main/java/com/usthe/warehouse/entrance/KafkaMetricsDataDeserializer.java
deleted file mode 100644
index 544265a..0000000
--- a/warehouse/src/main/java/com/usthe/warehouse/entrance/KafkaMetricsDataDeserializer.java
+++ /dev/null
@@ -1,24 +0,0 @@
-package com.usthe.warehouse.entrance;
-
-import com.usthe.common.entity.message.CollectRep;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.common.serialization.Deserializer;
-
-/**
- * MetricsData的反序列化
- * @author tom
- * @date 2021/11/24 17:29
- */
-@Slf4j
-public class KafkaMetricsDataDeserializer implements Deserializer {
-
- @Override
- public CollectRep.MetricsData deserialize(String topicName, byte[] bytes) {
- try {
- return CollectRep.MetricsData.parseFrom(bytes);
- } catch (Exception e) {
- log.error(e.getMessage(), e);
- }
- return null;
- }
-}
diff --git a/warehouse/src/main/java/com/usthe/warehouse/store/InfluxdbDataStorage.java b/warehouse/src/main/java/com/usthe/warehouse/store/InfluxdbDataStorage.java
deleted file mode 100644
index 22061a3..0000000
--- a/warehouse/src/main/java/com/usthe/warehouse/store/InfluxdbDataStorage.java
+++ /dev/null
@@ -1,134 +0,0 @@
-package com.usthe.warehouse.store;
-
-import com.influxdb.client.InfluxDBClient;
-import com.influxdb.client.InfluxDBClientFactory;
-import com.influxdb.client.WriteApi;
-import com.influxdb.client.WriteOptions;
-import com.influxdb.client.domain.WritePrecision;
-import com.influxdb.client.write.Point;
-import com.usthe.common.entity.message.CollectRep;
-import com.usthe.common.util.CommonConstants;
-import com.usthe.warehouse.MetricsDataQueue;
-import com.usthe.warehouse.WarehouseProperties;
-import com.usthe.warehouse.WarehouseWorkerPool;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.DisposableBean;
-import org.springframework.boot.autoconfigure.AutoConfigureAfter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Configuration;
-
-import java.time.Instant;
-import java.util.List;
-
-/**
- * influxdb存储采集数据
- * @author tom
- * @date 2021/11/24 18:23
- */
-@Configuration
-@AutoConfigureAfter(value = {WarehouseProperties.class})
-@ConditionalOnProperty(prefix = "warehouse.store.influxdb",
- name = "enabled", havingValue = "true", matchIfMissing = true)
-@Slf4j
-public class InfluxdbDataStorage implements DisposableBean {
-
- private InfluxDBClient influxClient;
- private WriteApi writeApi;
- private WarehouseWorkerPool workerPool;
- private MetricsDataQueue dataQueue;
-
- public InfluxdbDataStorage (WarehouseProperties properties, WarehouseWorkerPool workerPool,
- MetricsDataQueue dataQueue) {
- this.workerPool = workerPool;
- this.dataQueue = dataQueue;
- initInfluxDbClient(properties);
- startStorageData();
- }
-
- private void startStorageData() {
- Runnable runnable = () -> {
- Thread.currentThread().setName("warehouse-influxdb-data-storage");
- while (!Thread.currentThread().isInterrupted()) {
- try {
- CollectRep.MetricsData metricsData = dataQueue.pollInfluxMetricsData();
- if (metricsData != null) {
- saveData(metricsData);
- }
- } catch (InterruptedException e) {
- log.error(e.getMessage());
- }
- }
- };
- workerPool.executeJob(runnable);
- workerPool.executeJob(runnable);
- }
-
- private void initInfluxDbClient(WarehouseProperties properties) {
- if (properties == null || properties.getStore() == null || properties.getStore().getInfluxdb() == null) {
- log.error("init error, please config Warehouse influxdb props in application.yml");
- throw new IllegalArgumentException("please config Warehouse influxdb props");
- }
- WarehouseProperties.StoreProperties.InfluxdbProperties influxdbProp = properties.getStore().getInfluxdb();
- influxClient = InfluxDBClientFactory.create(influxdbProp.getServers(), influxdbProp.getToken().toCharArray(),
- influxdbProp.getOrg(), influxdbProp.getBucket());
- WriteOptions writeOptions = WriteOptions.builder()
- .batchSize(1000)
- .bufferLimit(1000)
- .jitterInterval(1000)
- .retryInterval(5000)
- .build();
- writeApi = influxClient.makeWriteApi(writeOptions);
- }
-
-
- public void saveData(CollectRep.MetricsData metricsData) {
- String measurement = metricsData.getApp() + "_" + metricsData.getMetrics();
- String monitorId = String.valueOf(metricsData.getId());
- Instant collectTime = Instant.ofEpochMilli(metricsData.getTime());
-
- List fields = metricsData.getFieldsList();
- for (CollectRep.ValueRow valueRow : metricsData.getValuesList()) {
- Point point = Point.measurement(measurement)
- .addTag("id", monitorId)
- .addTag("instance", valueRow.getInstance())
- .time(collectTime, WritePrecision.MS);
- for (int index = 0; index < fields.size(); index++) {
- CollectRep.Field field = fields.get(index);
- String value = valueRow.getColumns(index);
- if (field.getType() == CommonConstants.TYPE_NUMBER) {
- // number data
- if (CommonConstants.NULL_VALUE.equals(value)) {
- point.addField(field.getName(), (Number) null);
- } else {
- try {
- double number = Double.parseDouble(value);
- point.addField(field.getName(), number);
- } catch (Exception e) {
- log.warn(e.getMessage());
- point.addField(field.getName(), (Number) null);
- }
- }
- } else {
- // string
- if (CommonConstants.NULL_VALUE.equals(value)) {
- point.addField(field.getName(), (String) null);
- } else {
- point.addField(field.getName(), value);
- }
- }
- }
- writeApi.writePoint(point);
- }
- }
-
-
- @Override
- public void destroy() throws Exception {
- if (writeApi != null) {
- writeApi.close();
- }
- if (influxClient != null) {
- influxClient.close();
- }
- }
-}
diff --git a/warehouse/src/main/java/com/usthe/warehouse/store/RedisDataStorage.java b/warehouse/src/main/java/com/usthe/warehouse/store/RedisDataStorage.java
index 0287697..10a3d4f 100644
--- a/warehouse/src/main/java/com/usthe/warehouse/store/RedisDataStorage.java
+++ b/warehouse/src/main/java/com/usthe/warehouse/store/RedisDataStorage.java
@@ -1,7 +1,7 @@
package com.usthe.warehouse.store;
+import com.usthe.collector.dispatch.export.MetricsDataExporter;
import com.usthe.common.entity.message.CollectRep;
-import com.usthe.warehouse.MetricsDataQueue;
import com.usthe.warehouse.WarehouseProperties;
import com.usthe.warehouse.WarehouseWorkerPool;
import io.lettuce.core.RedisClient;
@@ -34,13 +34,12 @@ public class RedisDataStorage implements DisposableBean {
private RedisClient redisClient;
private StatefulRedisConnection connection;
private WarehouseWorkerPool workerPool;
- private MetricsDataQueue dataQueue;
+ private MetricsDataExporter dataExporter;
public RedisDataStorage (WarehouseProperties properties, WarehouseWorkerPool workerPool,
- MetricsDataQueue dataQueue) {
+ MetricsDataExporter dataExporter) {
this.workerPool = workerPool;
- this.dataQueue = dataQueue;
-
+ this.dataExporter = dataExporter;
initRedisClient(properties);
startStorageData();
}
@@ -55,7 +54,7 @@ public class RedisDataStorage implements DisposableBean {
Thread.currentThread().setName("warehouse-redis-data-storage");
while (!Thread.currentThread().isInterrupted()) {
try {
- CollectRep.MetricsData metricsData = dataQueue.pollRedisMetricsData();
+ CollectRep.MetricsData metricsData = dataExporter.pollWarehouseRedisMetricsData();
if (metricsData != null) {
saveData(metricsData);
}
diff --git a/warehouse/src/main/resources/META-INF/spring.factories b/warehouse/src/main/resources/META-INF/spring.factories
index fc593ec..783832d 100644
--- a/warehouse/src/main/resources/META-INF/spring.factories
+++ b/warehouse/src/main/resources/META-INF/spring.factories
@@ -2,7 +2,5 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.usthe.warehouse.WarehouseProperties,\
com.usthe.warehouse.MetricsDataQueue,\
com.usthe.warehouse.WarehouseWorkerPool,\
-com.usthe.warehouse.entrance.KafkaDataConsume,\
-com.usthe.warehouse.store.InfluxdbDataStorage,\
com.usthe.warehouse.store.RedisDataStorage,\
com.usthe.warehouse.controller.MetricsDataController
\ No newline at end of file