Java中的多线程竞争与CountDownLatch的施用
Java中的多线程竞争与CountDownLatch的应用
线程间的竞争叫做Racing,正如这篇文章的图片一样,不同的线程好比在高速赛道上行驶的F1方程式赛车,当赛车在各自的赛道上互不影响各自行驶时,可以相安无事。但这是一场比赛,既然是比赛,必然有并线、超车的行为,此时两辆赛车很可能会挤到赛道中公共的部分,如果此时两车速度相差不多,而且距离差距不大,就极有可能发生可怕的事故。
在计算机世界里,不同的线程运行的程序就好比不同赛道上的赛车。如果各个线程相互独立,完全没有关系,不会相互影响。但实际情况往往没有这么简单,线程可能会共同访问某个共享资源,并操作这些共享资源,如果程序逻辑依赖于这些资源,那么不同线程执行的先后次序就很有可能会影响到程序的执行结果。如果我们不对线程间的访问加以协调和控制,就可能导致错误的输出。
下面这个例子里,我想展示一下线程竞争的情况:
上面的这段代码是一个基础的Socket服务端。这个类侦听8080端口,并通过accept()方法来接收客户端的请求,当客户端有连接时,返回给客户端一个"."的字符。此外,这个类中还包含一个close()方法,用于结束服务。整个代码最关键的就是running这个变量,通过设置running为true,服务器保持接受请求进行服务,如果服务被 close掉,running就被置为false ,此时客户端如果有请求,服务器就会输出一个"G",表示服务已关闭。如果在running的过程中出现Exception,则像屏幕输出一个"C",并直接返回。
一切看起来都完美无缺,但真的如此吗?我们来写一个测试程序来使用这段代码,看看它是否像我们期望的那样工作。首先我们撰写一个线程类,用于启动上面的服务端代码:
这段代码的作用是把服务端的代码封装进一个Thread,让服务端在独立的线程中运行,防止我们要撰写的主程序被服务端的accept()给block住。并向屏幕输出一个"A"表示服务启动。
接下来我们撰写客户端程序,让客户端连接至8080端口,持续不断地访问服务端,并向屏幕打印服务端的返回数据:
最后是主程序:
把主程序跑起来,我们得到的结果如下:
出乎意科的事情发生了,主程序执行了server.close();后,服务端程序没有正确地退出,而是在accept()方法里抛出了异常,打印出了C的结果,而不是期望中的G。这是为什么呢?其实原因出在running这个变量上。当主程序发送了server.close();的请求后,running被置为false, 但客户端的速度也是很快的,此时可能已经发起请求,并已经由服务端进行服务。但服务端此时恰好已经关闭了socket,因此在服务端的主程序中抛出了异常IOException, 并打印了字符 C。
这就是一个典型的线程竞争的场景,该如何解决这个问题呢?其实我们的逻辑没有错,只不过在上面的代码里,对running的写操作和判断running的状态没有协调好。我们的代码应该保证在running设置为false后,所有当前在跑的客户端程序都结束后,再关闭socket。 通过使用import java.util.concurrent.CountDownLatch, 我们便可以实现这一点:
如上面的代码所示,我们首先创建了一个计数锁:
在accept方法里面,我们在running被置为false后,将锁释放掉:
然后在close方法里,等待这个锁释放后,才可以将socket关闭:
注意上面代码中的逻辑顺序,首先将running置为false,然后等待accept方法中逻辑执行完后将锁释放掉,然后才可以继续向下执行socket.close()方法关闭连接。如果accept方法没有释放锁,close方法中的lock.await()将一直处于等待状态,从而避免过早关闭socket。
重新运行测试代码,程序输出如下:
可以看到,程序已经按照我们预期来工作。
我将上面用到的例子放在了github上面,有兴趣可以下载下来玩玩看:
下载完成后,首先需要编译代码:
然后可运行maven命令运行例子:
上面的命令将会使用SampleServerWithRacingProblem产生线程竞争的结果。下面的例子则使用SampleServerWithoutRacingProblem,由于锁的保护,不会产生竞争:
线程间的竞争叫做Racing,正如这篇文章的图片一样,不同的线程好比在高速赛道上行驶的F1方程式赛车,当赛车在各自的赛道上互不影响各自行驶时,可以相安无事。但这是一场比赛,既然是比赛,必然有并线、超车的行为,此时两辆赛车很可能会挤到赛道中公共的部分,如果此时两车速度相差不多,而且距离差距不大,就极有可能发生可怕的事故。
在计算机世界里,不同的线程运行的程序就好比不同赛道上的赛车。如果各个线程相互独立,完全没有关系,不会相互影响。但实际情况往往没有这么简单,线程可能会共同访问某个共享资源,并操作这些共享资源,如果程序逻辑依赖于这些资源,那么不同线程执行的先后次序就很有可能会影响到程序的执行结果。如果我们不对线程间的访问加以协调和控制,就可能导致错误的输出。
下面这个例子里,我想展示一下线程竞争的情况:
package net.bluedash.countdownlatch; import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; public class SampleServerWithRacingProblem { private boolean running = true; private ServerSocket socket; public SampleServerWithRacingProblem() throws IOException { init(); } private void init() throws IOException { socket = new ServerSocket(8080); } public void accept() { while (running) { try { Socket connection = socket.accept(); connection.getOutputStream().write(". ".getBytes()); connection.getOutputStream().flush(); connection.close(); } catch (IOException e) { System.out.print("\nC"); return; } } System.out.print("\nG"); } public void close() throws IOException, InterruptedException { running = false; socket.close(); }
上面的这段代码是一个基础的Socket服务端。这个类侦听8080端口,并通过accept()方法来接收客户端的请求,当客户端有连接时,返回给客户端一个"."的字符。此外,这个类中还包含一个close()方法,用于结束服务。整个代码最关键的就是running这个变量,通过设置running为true,服务器保持接受请求进行服务,如果服务被 close掉,running就被置为false ,此时客户端如果有请求,服务器就会输出一个"G",表示服务已关闭。如果在running的过程中出现Exception,则像屏幕输出一个"C",并直接返回。
一切看起来都完美无缺,但真的如此吗?我们来写一个测试程序来使用这段代码,看看它是否像我们期望的那样工作。首先我们撰写一个线程类,用于启动上面的服务端代码:
package net.bluedash.countdownlatch; public class ServerWorker implements Runnable { private SampleServerWithRacingProblem server; public ServerWorker(SampleServerWithRacingProblem server) { this.server = server; } @Override public void run() { System.out.print("A"); server.accept(); } }
这段代码的作用是把服务端的代码封装进一个Thread,让服务端在独立的线程中运行,防止我们要撰写的主程序被服务端的accept()给block住。并向屏幕输出一个"A"表示服务启动。
接下来我们撰写客户端程序,让客户端连接至8080端口,持续不断地访问服务端,并向屏幕打印服务端的返回数据:
package net.bluedash.countdownlatch; import java.io.BufferedReader; import java.io.InputStreamReader; import java.net.Socket; public class ClientWorker implements Runnable { @Override public void run() { try { while (true) { Thread.sleep(100); // 防止客户端请求太快,消耗大量系统资源 Socket socket = new Socket("127.0.0.1", 8080); BufferedReader rd = new BufferedReader(new InputStreamReader( socket.getInputStream())); String line; while ((line = rd.readLine()) != null) { System.out.print(line); } socket.close(); } } catch (Exception ignore) { } } }
最后是主程序:
package net.bluedash.countdownlatch; import java.io.IOException; public class UseSampleServer { public static void main(String[] args) throws IOException, InterruptedException { SampleServerWithRacingProblem server = new SampleServerWithRacingProblem(); ServerWorker serverWorker = new ServerWorker(server); Thread serverThread = new Thread(serverWorker); serverThread.start(); // 开启5个客户端连接 for (int i = 0; i < 5; i++) { ClientWorker clientWorker = new ClientWorker(); Thread clientThread = new Thread(clientWorker); clientThread.start(); } // 给客户端时间5秒钟时间去跑起来,防止服务器过早地关闭 Thread.currentThread().sleep(5000); server.close(); System.out.print("\nQ"); } }
把主程序跑起来,我们得到的结果如下:
A. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . Q C
出乎意科的事情发生了,主程序执行了server.close();后,服务端程序没有正确地退出,而是在accept()方法里抛出了异常,打印出了C的结果,而不是期望中的G。这是为什么呢?其实原因出在running这个变量上。当主程序发送了server.close();的请求后,running被置为false, 但客户端的速度也是很快的,此时可能已经发起请求,并已经由服务端进行服务。但服务端此时恰好已经关闭了socket,因此在服务端的主程序中抛出了异常IOException, 并打印了字符 C。
这就是一个典型的线程竞争的场景,该如何解决这个问题呢?其实我们的逻辑没有错,只不过在上面的代码里,对running的写操作和判断running的状态没有协调好。我们的代码应该保证在running设置为false后,所有当前在跑的客户端程序都结束后,再关闭socket。 通过使用import java.util.concurrent.CountDownLatch, 我们便可以实现这一点:
package net.bluedash.countdownlatch; import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; import java.util.concurrent.CountDownLatch; public class SampleServerWithoutRacingProblem { private boolean running = true; private ServerSocket socket; private CountDownLatch lock = new CountDownLatch(1); public SampleServerWithoutRacingProblem() throws IOException { init(); } private void init() throws IOException { socket = new ServerSocket(8080); } public void accept() { while (running) { try { Socket connection = socket.accept(); connection.getOutputStream().write(". ".getBytes()); connection.getOutputStream().flush(); connection.close(); } catch (IOException e) { System.out.print("\nC"); return; } } lock.countDown(); System.out.print("\nG"); } public void close() throws IOException, InterruptedException { running = false; lock.await(); socket.close(); } }
如上面的代码所示,我们首先创建了一个计数锁:
private CountDownLatch lock = new CountDownLatch(1);
在accept方法里面,我们在running被置为false后,将锁释放掉:
lock.countDown();
然后在close方法里,等待这个锁释放后,才可以将socket关闭:
public void close() throws IOException, InterruptedException { running = false; lock.await(); socket.close(); }
注意上面代码中的逻辑顺序,首先将running置为false,然后等待accept方法中逻辑执行完后将锁释放掉,然后才可以继续向下执行socket.close()方法关闭连接。如果accept方法没有释放锁,close方法中的lock.await()将一直处于等待状态,从而避免过早关闭socket。
重新运行测试代码,程序输出如下:
A. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . G Q.
可以看到,程序已经按照我们预期来工作。
我将上面用到的例子放在了github上面,有兴趣可以下载下来玩玩看:
git clone git://github.com/liweinan/try-countdownlatch.git
下载完成后,首先需要编译代码:
mvn install
然后可运行maven命令运行例子:
mvn exec:java -Dexec.mainClass="net.bluedash.countdownlatch.UseSampleServer" -Dexec.args='WithProblem'
上面的命令将会使用SampleServerWithRacingProblem产生线程竞争的结果。下面的例子则使用SampleServerWithoutRacingProblem,由于锁的保护,不会产生竞争:
mvn exec:java -Dexec.mainClass="net.bluedash.countdownlatch.UseSampleServer" -Dexec.args='WithoutProblem'