worker

im-entry/src/main/java/net/sopod/soim/entry/worker/TaskEvent.java

@@ -0,0 +1,13 @@
+package net.sopod.soim.entry.worker;
+
+public class TaskEvent {
+    private Runnable task;
+
+    public Runnable getTask() {
+        return task;
+    }
+
+    public void setTask(Runnable task) {
+        this.task = task;
+    }
+}

im-entry/src/main/java/net/sopod/soim/entry/worker/Worker.java

@@ -0,0 +1,72 @@
+package net.sopod.soim.entry.worker;
+
+import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.RingBuffer;
+import com.lmax.disruptor.YieldingWaitStrategy;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.dsl.ProducerType;
+import net.sopod.soim.common.util.ImClock;
+import net.sopod.soim.entry.util.FastThreadLocalThreadFactory;
+
+import java.util.concurrent.*;
+
+/**
+ * dispatch -> worker * core_num -> disruptor 队列执行
+ */
+public class Worker implements EventHandler<TaskEvent>, EventFactory<TaskEvent> {
+
+    private Disruptor<TaskEvent> disruptor;
+
+    private RingBuffer<TaskEvent> ringBuffer;
+
+    private ExecutorService executor;
+
+    public Worker(String workerThread) {
+        // Executors.newSingleThreadExecutor(new FastThreadLocalThreadFactory(workerThread, Thread.MAX_PRIORITY));
+        this.executor = new ThreadPoolExecutor(
+                1,
+                1,
+                0,
+                TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<Runnable>(),
+                new FastThreadLocalThreadFactory(workerThread, Thread.MAX_PRIORITY)
+        );
+        this.disruptor = new Disruptor<>(
+                this,
+                16 * 1024,
+                executor,
+                ProducerType.SINGLE,
+                new YieldingWaitStrategy());
+        this.disruptor.handleEventsWith(this);
+        this.disruptor.start();
+        this.ringBuffer = disruptor.getRingBuffer();
+    }
+
+    public void execute(Runnable runnable) {
+        long next = ringBuffer.next();
+        TaskEvent taskEvent = ringBuffer.get(next);
+        taskEvent.setTask(runnable);
+        ringBuffer.publish(next);
+    }
+
+    public void shutdown() {
+        this.disruptor.shutdown();
+        this.executor.shutdown();
+    }
+
+    @Override
+    public void onEvent(TaskEvent taskEvent, long sequence, boolean endOfBatch) throws Exception {
+        long start = ImClock.millis();
+        taskEvent.getTask().run();
+        long time = ImClock.millis() - start;
+        if (time > 100) {
+            System.out.println("任务执行时间过长:" + time);
+        }
+    }
+
+    @Override
+    public TaskEvent newInstance() {
+        return new TaskEvent();
+    }
+}

im-entry/src/main/java/net/sopod/soim/entry/worker/WorkerGroup.java

@@ -0,0 +1,52 @@
+package net.sopod.soim.entry.worker;
+
+import com.google.protobuf.GeneratedMessageV3;
+import net.sopod.soim.core.session.Account;
+import net.sopod.soim.core.session.NetUser;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * dispatch -> worker * core_num -> disruptor 队列执行
+ */
+public class WorkerGroup {
+
+    public static final int MAX_WORKER_SIZE = 16;
+
+    private static Worker[] WORKERS;
+
+    private static AtomicInteger counter;
+
+    public static void init(int codeSize) {
+        if (WORKERS != null) {
+            return;
+        }
+        WORKERS = new Worker[codeSize];
+        for (int i = 0; i < codeSize; i++) {
+            WORKERS[i] = new Worker("group-worker-" + i);
+        }
+        counter = new AtomicInteger(-1);
+    }
+
+    public static Worker next() {
+        counter.compareAndSet(Integer.MAX_VALUE, -1);
+        return WORKERS[counter.incrementAndGet() % WORKERS.length];
+    }
+
+    public static void publish(NetUser netUser, GeneratedMessageV3 message) {
+        next().execute(() -> {
+
+        });
+    }
+
+    public static void publish(Account netUser, GeneratedMessageV3 message) {
+
+    }
+
+    public static void shutdown() {
+        for (Worker worker : WORKERS) {
+            worker.shutdown();
+        }
+    }
+
+}