java - Java Scheduled Executor 线程进入等待状态
问题描述
因此，我正在尝试使用两个自定义线程和一个预定的线程执行器来实现从我的 Java 客户端到 Go 服务器的基本异步消息传递协议。因此，我启动了一个从 Java 客户端到 Go 服务器的 TCP 套接字，并在两侧设置了读取和写入缓冲区。
Writer 线程将每个请求对应的唯一 ID 保存到 Redis 服务器中，并在 TCP 套接字上使用 PrintWriter 写入请求。
作家线程
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 中的唯一 ID，并将它们标记为完成（成功/失败）。
读者线程
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 服务器不是问题，否则单线程执行器也应该卡住，但事实并非如此。可能，我遇到了一些僵局，但无法找出确切的原因。有人可以帮忙吗。如果你还想要什么，请告诉我。
解决方案
代码的套接字和流部分似乎是正确的。并行使用来自套接字的输入和输出流是正确的，因为它们是独立的流。
问题是 Jedis 组件。Jedis 不是线程安全的。Jedis 文档说：
您不应该使用来自不同线程的相同实例，因为您会遇到奇怪的错误。有时创建大量 Jedis 实例还不够好，因为这意味着大量的套接字和连接，这也会导致奇怪的错误。单个 Jedis 实例不是线程安全的！...
需要重写代码，使每个线程都有一个私有的 Jedis 实例。
推荐阅读
- android - LinearLayout 被用作按钮卡在按下状态
- r - 通过逐行应用函数并创建（可能）更大的列来转换 R 数据框
- javascript - 标记的 onLoad 事件
不会在初始页面访问时触发
- python - jinja2 '{% set ... %}' 在 ansible playbook 中的处理
- java - 是否可以检查字符串数组，如果其中的元素可以转换为 int 或 double？
- android - 当 Listview 中的项目溢出时，我的应用程序崩溃
- r - MacOSX 中的 XRJulia 和 findJulia
- sql-server - 如何将表从 SQL Linked Server 复制到 SQL Server 并维护关键约束？
- angular - 在自定义结构指令Angular 6+中使用NGRX Observable
- c++ - QT eventFilter mouseEvent->pos().x() always zero but mouseEvent->pos().y() working fine