Потоки Java Scheduled Executor, переходящие в состояние ожидания - PullRequest
Новая
       11

Потоки Java Scheduled Executor, переходящие в состояние ожидания

0 голосов
/ 21 мая 2018

Итак, я пытаюсь реализовать базовый протокол асинхронного обмена сообщениями с моего Java-клиента на Go-сервер, используя два пользовательских потока, с запланированным исполнителем потока.Итак, я инициирую один сокет TCP с Java-клиента на Go-сервер и устанавливаю буферы чтения и записи с обеих сторон.

Поток модуля записи сохраняет уникальный идентификатор на сервере Redis, соответствующий каждому запросу, и записываю запрос, используяPrintWriter на TCP-сокете.

WriterThread 

public class WithdrawRequestWriter extends Thread {

    private static final int START_ACCOUNT_NO = 1;
    private static final int END_ACCOUNT_NO = 2_000;
    private static final int AMOUNT_TO_WITHDRAW = 100;

    private PrintWriter writer;
    private Jedis jedis;

    public WithdrawRequestWriter(PrintWriter writer, Jedis jedis) {
        this.writer = writer;
        this.jedis = jedis;
    }

    @Override
    public void run() {
        int i = 0;
        for (int accountNo = START_ACCOUNT_NO; accountNo <= END_ACCOUNT_NO; accountNo++) {
            String uniqueId = UUID.randomUUID().toString();
            if (jedis.set(uniqueId, String.valueOf(accountNo), "NX", "EX", 10) != null) {
                writer.write(uniqueId + " " + accountNo + " " + AMOUNT_TO_WITHDRAW + "\n");
                writer.flush();
            } else {
                System.err.println("Setting of UUID in redis failed: " + uniqueId);
            }
        }
    }
}

Аналогично, поток чтения ожидает ответа от сервера.Как только он получает его, он совпадает с уникальным идентификатором в Redis, он помечает их как выполненные (успех / неудача).

ReaderThread 

public class WithdrawRequestReader extends Thread {

    private BufferedReader reader;
    private Jedis jedis;


    public WithdrawRequestReader(BufferedReader reader, Jedis jedis) {
        this.reader = reader;
        this.jedis = jedis;
    }

    @Override
    public void run() {
        String response;
        while (true) {
            try {
                if ((response = reader.readLine()) != null) {
                    String[] tokenizedResponse = response.split(" ");
                    System.out.println("Response is: " + response);
                    String value = jedis.get(tokenizedResponse[0]);
                    if (value != null) {
                        System.out.println("Request Token: " + tokenizedResponse[0]
                                + " | Status: " + tokenizedResponse[1]
                                + " | Account: " + tokenizedResponse[2]
                                + " | Amount: " + tokenizedResponse[3]
                                + " | Remaining Balance: " + tokenizedResponse[4]
                                + " | ResponeStr: " + tokenizedResponse[5]);
                    }
                } else {
                    System.out.println("Waiting for response...");
                }
            } catch (IOException ex) {
                ex.printStackTrace();
                System.exit(3);
            }
        }
    }
}

Основной поток 

public class TestClient {

    private static final String HOSTNAME = "localhost";
    private static final int PORT = 12410;

    public static void main(String[] args) throws IOException {

        Socket socket = new Socket(HOSTNAME, PORT);
        PrintWriter writer = new PrintWriter(socket.getOutputStream());
        BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
        Jedis jedis = new Jedis("localhost");

        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
//        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        WithdrawRequestReader readerThread = new WithdrawRequestReader(reader, jedis);
        WithdrawRequestWriter writerThread = new WithdrawRequestWriter(writer, jedis);
        scheduledExecutorService.schedule(writerThread, 0, TimeUnit.MILLISECONDS);
        scheduledExecutorService.schedule(readerThread, 0, TimeUnit.MILLISECONDS);
    }
}

Когда я использую singlePoolThreadExecutor, задачи выполняются последовательно (как и ожидалось), и все идет гладко.Но когда я использую threadPoolExecutor с двумя потоками, после нескольких операций чтения и записи они переходят в состояние ожидания на неопределенное время.Я проверил из бесплатных инструментов, доступных онлайн, есть ли тупик, но мне не помогли.Ниже дамп потока, когда они застряли.

Дамп потока 

Full thread dump Java HotSpot(TM) 64-Bit Server VM (9.0.4+11 mixed mode):

"DestroyJavaVM" #16 prio=5 os_prio=31 tid=0x00007f974e06d000 nid=0x2503 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

"pool-1-thread-2" #15 prio=5 os_prio=31 tid=0x00007f974e06c000 nid=0x6003 waiting on condition [0x000070000c819000]
   java.lang.Thread.State: WAITING (parking)
    at jdk.internal.misc.Unsafe.park(java.base@9.0.4/Native Method)
    - parking to wait for  <0x0000000747cf64b0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
    at java.util.concurrent.locks.LockSupport.park(java.base@9.0.4/LockSupport.java:194)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(java.base@9.0.4/AbstractQueuedSynchronizer.java:2062)
    at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(java.base@9.0.4/ScheduledThreadPoolExecutor.java:1119)
    at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(java.base@9.0.4/ScheduledThreadPoolExecutor.java:848)
    at java.util.concurrent.ThreadPoolExecutor.getTask(java.base@9.0.4/ThreadPoolExecutor.java:1092)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@9.0.4/ThreadPoolExecutor.java:1152)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@9.0.4/ThreadPoolExecutor.java:641)
    at java.lang.Thread.run(java.base@9.0.4/Thread.java:844)

"pool-1-thread-1" #14 prio=5 os_prio=31 tid=0x00007f974e8a9800 nid=0x5e03 waiting on condition [0x000070000c716000]
   java.lang.Thread.State: WAITING (parking)
    at jdk.internal.misc.Unsafe.park(java.base@9.0.4/Native Method)
    - parking to wait for  <0x0000000747cf64b0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
    at java.util.concurrent.locks.LockSupport.park(java.base@9.0.4/LockSupport.java:194)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(java.base@9.0.4/AbstractQueuedSynchronizer.java:2062)
    at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(java.base@9.0.4/ScheduledThreadPoolExecutor.java:1119)
    at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(java.base@9.0.4/ScheduledThreadPoolExecutor.java:848)
    at java.util.concurrent.ThreadPoolExecutor.getTask(java.base@9.0.4/ThreadPoolExecutor.java:1092)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@9.0.4/ThreadPoolExecutor.java:1152)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@9.0.4/ThreadPoolExecutor.java:641)
    at java.lang.Thread.run(java.base@9.0.4/Thread.java:844)

"Service Thread" #11 daemon prio=9 os_prio=31 tid=0x00007f974d827000 nid=0xa003 runnable [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

"Monitor Ctrl-Break" #10 daemon prio=5 os_prio=31 tid=0x00007f974e078800 nid=0x5903 runnable [0x000070000c40d000]
   java.lang.Thread.State: RUNNABLE
    at java.net.SocketInputStream.socketRead0(java.base@9.0.4/Native Method)
    at java.net.SocketInputStream.socketRead(java.base@9.0.4/SocketInputStream.java:116)
    at java.net.SocketInputStream.read(java.base@9.0.4/SocketInputStream.java:171)
    at java.net.SocketInputStream.read(java.base@9.0.4/SocketInputStream.java:141)
    at sun.nio.cs.StreamDecoder.readBytes(java.base@9.0.4/StreamDecoder.java:284)
    at sun.nio.cs.StreamDecoder.implRead(java.base@9.0.4/StreamDecoder.java:326)
    at sun.nio.cs.StreamDecoder.read(java.base@9.0.4/StreamDecoder.java:178)
    - locked <0x0000000747d9ea78> (a java.io.InputStreamReader)
    at java.io.InputStreamReader.read(java.base@9.0.4/InputStreamReader.java:185)
    at java.io.BufferedReader.fill(java.base@9.0.4/BufferedReader.java:161)
    at java.io.BufferedReader.readLine(java.base@9.0.4/BufferedReader.java:326)
    - locked <0x0000000747d9ea78> (a java.io.InputStreamReader)
    at java.io.BufferedReader.readLine(java.base@9.0.4/BufferedReader.java:392)
    at com.intellij.rt.execution.application.AppMainV2$1.run(AppMainV2.java:64)

"Common-Cleaner" #9 daemon prio=8 os_prio=31 tid=0x00007f974e00a000 nid=0xa203 in Object.wait() [0x000070000c30a000]
   java.lang.Thread.State: TIMED_WAITING (on object monitor)
    at java.lang.Object.wait(java.base@9.0.4/Native Method)
    - waiting on <0x0000000747f37610> (a java.lang.ref.ReferenceQueue$Lock)
    at java.lang.ref.ReferenceQueue.remove(java.base@9.0.4/ReferenceQueue.java:151)
    - waiting to re-lock in wait() <0x0000000747f37610> (a java.lang.ref.ReferenceQueue$Lock)
    at jdk.internal.ref.CleanerImpl.run(java.base@9.0.4/CleanerImpl.java:148)
    at java.lang.Thread.run(java.base@9.0.4/Thread.java:844)
    at jdk.internal.misc.InnocuousThread.run(java.base@9.0.4/InnocuousThread.java:122)

"Sweeper thread" #8 daemon prio=9 os_prio=31 tid=0x00007f974d81c000 nid=0x5803 runnable [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

"C1 CompilerThread2" #7 daemon prio=9 os_prio=31 tid=0x00007f974d81b000 nid=0xa503 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE
   No compile task

"C2 CompilerThread1" #6 daemon prio=9 os_prio=31 tid=0x00007f974e005000 nid=0xa603 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE
   No compile task

"C2 CompilerThread0" #5 daemon prio=9 os_prio=31 tid=0x00007f974d81a800 nid=0xa803 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE
   No compile task

"Signal Dispatcher" #4 daemon prio=9 os_prio=31 tid=0x00007f974e01a800 nid=0x5603 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

"Finalizer" #3 daemon prio=8 os_prio=31 tid=0x00007f974d041800 nid=0x4003 in Object.wait() [0x000070000bc75000]
   java.lang.Thread.State: WAITING (on object monitor)
    at java.lang.Object.wait(java.base@9.0.4/Native Method)
    - waiting on <0x0000000747f0d078> (a java.lang.ref.ReferenceQueue$Lock)
    at java.lang.ref.ReferenceQueue.remove(java.base@9.0.4/ReferenceQueue.java:151)
    - waiting to re-lock in wait() <0x0000000747f0d078> (a java.lang.ref.ReferenceQueue$Lock)
    at java.lang.ref.ReferenceQueue.remove(java.base@9.0.4/ReferenceQueue.java:172)
    at java.lang.ref.Finalizer$FinalizerThread.run(java.base@9.0.4/Finalizer.java:216)

"Reference Handler" #2 daemon prio=10 os_prio=31 tid=0x00007f974d040800 nid=0x4103 waiting on condition [0x000070000bb72000]
   java.lang.Thread.State: RUNNABLE
    at java.lang.ref.Reference.waitForReferencePendingList(java.base@9.0.4/Native Method)
    at java.lang.ref.Reference.processPendingReferences(java.base@9.0.4/Reference.java:174)
    at java.lang.ref.Reference.access$000(java.base@9.0.4/Reference.java:44)
    at java.lang.ref.Reference$ReferenceHandler.run(java.base@9.0.4/Reference.java:138)

"VM Thread" os_prio=31 tid=0x00007f974e884000 nid=0x4303 runnable 

"GC Thread#0" os_prio=31 tid=0x00007f974e80d800 nid=0x2a03 runnable 

"GC Thread#1" os_prio=31 tid=0x00007f974e80e800 nid=0x5303 runnable 

"GC Thread#2" os_prio=31 tid=0x00007f974d80c800 nid=0x2b03 runnable 

"GC Thread#3" os_prio=31 tid=0x00007f974d000800 nid=0x2c03 runnable 

"G1 Main Marker" os_prio=31 tid=0x00007f974e835800 nid=0x4803 runnable 

"G1 Marker#0" os_prio=31 tid=0x00007f974d80d800 nid=0x4503 runnable 

"G1 Refine#0" os_prio=31 tid=0x00007f974e80f800 nid=0x4a03 runnable 

"G1 Refine#1" os_prio=31 tid=0x00007f974d80d000 nid=0x4c03 runnable 

"G1 Refine#2" os_prio=31 tid=0x00007f974e80f000 nid=0x4e03 runnable 

"G1 Refine#3" os_prio=31 tid=0x00007f974e806800 nid=0x5003 runnable 

"G1 Young RemSet Sampling" os_prio=31 tid=0x00007f974e810800 nid=0x2f03 runnable 

"VM Periodic Task Thread" os_prio=31 tid=0x00007f974e079800 nid=0x5c03 waiting on condition 

JNI global references: 421

Heap
 garbage-first heap   total 131072K, used 10240K [0x0000000740000000, 0x0000000740100400, 0x00000007c0000000)
  region size 1024K, 11 young (11264K), 0 survivors (0K)
 Metaspace       used 8824K, capacity 9274K, committed 9472K, reserved 1058816K
  class space    used 832K, capacity 927K, committed 1024K, reserved 1048576K

Я считаю, что сервер Go не является проблемой, в противном случае исполнитель одного потока должен такжезастрял, но это не так.Возможно, у меня тупик, но я не могу точно определить причину.Может кто-нибудь, пожалуйста, помогите.Если вы хотите что-нибудь еще, пожалуйста, дайте мне знать.

1 Ответ

0 голосов
/ 22 мая 2018

Сокет и потоковая часть кода кажутся правильными.Правильно использовать входной и выходной потоки из сокета параллельно, поскольку они являются независимыми потоками.

Проблема в компоненте Jedis.Джедай не нить безопасен.Документация Jedis гласит:

Вы не должны использовать один и тот же экземпляр из разных потоков, потому что у вас будут странные ошибки.И иногда создание большого количества экземпляров Jedis не достаточно хорошо, потому что это означает множество сокетов и соединений, что также приводит к странным ошибкам.Один экземпляр Jedis не является потокобезопасным! ...

Необходимо переписать код, чтобы у каждого потока был личный экземпляр Jedis.

...