_坑爹啊，fastdfs的并发问题

前言

使用了fastdfs&＃xff0c;由于要控制用户读取图片文件的权限&＃xff0c;所以没有用nginx的读取方式&＃xff0c;而是用java的downloadfile来获取。
结果。。。
同时读取四张图片90%有一张读不出来&＃xff0c;
会报错&＃xff1a;
如下&＃xff1a;

好了&＃xff0c;请参考&＃xff1a;
fastDFS遇到的并发问题recv cmd: 0 is not correct, expect cmd: 100

那一定都是你的错&＃xff01;- 一次FastDFS并发问题的排查经历

此贴是我对一篇帖子FastDFS并发会有bug&＃xff0c;其实我也不太信&＃xff1f;进行的一些测试和我自己也发现的相关问题与解决方案。最开始我也按照帖子进行了测试&＃xff0c;结果发现确实存在并发问题&＃xff08;下面代码改为单线程是木有问题的&＃xff09;。以下列出我尝试的几种情况&＃xff1a;先列出我进行测试的有main函数的FastConcurrence类&＃xff1a;public class FastConcurrence {private static int poolSize&＃61;2;//定义线程个数public static void main(String[] args) throws InterruptedException {latchTest();}private static void latchTest() throws InterruptedException {final CountDownLatch start &＃61; new CountDownLatch(1);final CountDownLatch end &＃61; new CountDownLatch(poolSize);ExecutorService exce &＃61; Executors.newFixedThreadPool(poolSize);for (int i &＃61; 0; i }
①把trackerClient,trackerServer,storageServer,storageClient设为全局变量。在类加载的时候&＃xff0c;就进行了初始化&＃xff0c;关键代码如下&＃xff1a;private static TrackerClient trackerClient; private static TrackerServer trackerServer;private static StorageServer storageServer;private static StorageClient storageClient;static { try {// 初始化文件资源 ClientGlobal.init("C:\\Users\\jianbo\\Downloads\\FastDFS\\conf\\client.conf");trackerClient &＃61; new TrackerClient(); trackerServer &＃61; trackerClient.getConnection(); //有并发问题&＃xff0c;所以勿重用storageClientstorageClient &＃61; new StorageClient(trackerServer, storageServer);} catch (Exception e) { logger.error(logger, e);} }
果不其然报错了&＃xff1a;
java.net.SocketException: socket closedat java.net.SocketInputStream.socketRead0(Native Method)at java.net.SocketInputStream.socketRead(Unknown Source)at java.net.SocketInputStream.read(Unknown Source)at java.net.SocketInputStream.read(Unknown Source)at org.csource.fastdfs.ProtoCommon.recvPackage(ProtoCommon.java:263)at org.csource.fastdfs.TrackerClient.getStoreStorage(TrackerClient.java:143)at org.csource.fastdfs.StorageClient.newWritableStorageConnection(StorageClient.java:1938)at org.csource.fastdfs.StorageClient.do_upload_file(StorageClient.java:703)at org.csource.fastdfs.StorageClient.upload_file(StorageClient.java:208)at org.csource.fastdfs.StorageClient.upload_file(StorageClient.java:226)at fastdfs.FileManager.upload(FileManager.java:76)at fastdfs.FastConcurrence.testLoad(FastConcurrence.java:48)at fastdfs.FastConcurrence.access$0(FastConcurrence.java:42)at fastdfs.FastConcurrence$1.run(FastConcurrence.java:27)at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)at java.util.concurrent.FutureTask.run(Unknown Source)at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)at java.lang.Thread.run(Unknown Source)
java.io.IOException: recv cmd: 0 is not correct, expect cmd: 100at org.csource.fastdfs.ProtoCommon.recvHeader(ProtoCommon.java:219)at org.csource.fastdfs.ProtoCommon.recvPackage(ProtoCommon.java:250)at org.csource.fastdfs.TrackerClient.getStoreStorage(TrackerClient.java:143)at org.csource.fastdfs.StorageClient.newWritableStorageConnection(StorageClient.java:1938)at org.csource.fastdfs.StorageClient.do_upload_file(StorageClient.java:703)at org.csource.fastdfs.StorageClient.upload_file(StorageClient.java:208)at org.csource.fastdfs.StorageClient.upload_file(StorageClient.java:226)at fastdfs.FileManager.upload(FileManager.java:76)at fastdfs.FastConcurrence.testLoad(FastConcurrence.java:48)at fastdfs.FastConcurrence.access$0(FastConcurrence.java:42)at fastdfs.FastConcurrence$1.run(FastConcurrence.java:27)at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)at java.util.concurrent.FutureTask.run(Unknown Source)at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)at java.lang.Thread.run(Unknown Source)
②我在调用storageClient.upload_file(byte[] file_buff, String file_ext_name, NameValuePair[] meta_list)前一句再storageClient &＃61; new StorageClient(trackerServer, storageServer);
关键代码如下&＃xff1a;public static String[] upload(FastDFSFile file) {try {storageClient &＃61; new StorageClient(trackerServer, storageServer);//新加的uploadResults &＃61; storageClient.upload_file(file.getContent(), file.getExt(), meta_list);} catch (Exception e) {logger.error("Exception when uploadind the file:" &＃43; file.getName(), e);}//省略一部分代码return uploadResults;
}
好了&＃xff0c;出现了和上面①一样的问题。
我发现&＃xff0c;其实这样本质没有变&＃xff0c;还是去改了全局storageClient 。应该new个新的&＃xff0c;所以产生了③。③我在调用storageClient.upload_file(byte[] file_buff, String file_ext_name, NameValuePair[] meta_list)前一句再StorageClient storageClient &＃61; new StorageClient(trackerServer, storageServer); 就是原帖最后说明的解决办法。public static String[] upload(FastDFSFile file) {try {StorageClient storageClient &＃61; new StorageClient(trackerServer, storageServer);//新加的uploadResults &＃61; storageClient.upload_file(file.getContent(), file.getExt(), meta_list);} catch (Exception e) {logger.error("Exception when uploadind the file:" &＃43; file.getName(), e);}//省略一部分代码return uploadResults;
}
我还是报错&＃xff0c;和前面一样的异常。原因还没有来得及找&＃xff0c;不过我采用锁是没有问题的。
④在调用方法的时候采用锁就能解决。例如这个方法我加了锁就没有问题了。private synchronized static void testLoad() throws Exception {String filePath&＃61;"D:\\fastDFS\\ques.png";File content&＃61;new File(filePath);TestFileManager test&＃61;new TestFileManager();FastDFSFile file &＃61;test.getFastFile(content,"png");for (int i&＃61;0;i<10;i&＃43;&＃43;){FileManager.upload(file);}System.out.println("完成一个线程!");}
一路顺畅。不过③为什么报错了&＃xff0c;我还要研究下。补&＃xff1a;对于③我找到问题了。源码下载下来&＃xff0c;然后我就查StorageClient类&＃xff0c;发现在do_upload_file方法中&＃xff0c;有一段代码&＃xff1a;bUploadSlave &＃61; ((group_name !&＃61; null && group_name.length() > 0) && (master_filename !&＃61; null && master_filename.length() > 0) &&(prefix_name !&＃61; null));if (bUploadSlave){bNewConnection &＃61; this.newUpdatableStorageConnection(group_name, master_filename);}else{bNewConnection &＃61; this.newWritableStorageConnection(group_name);}
其中的
this.newWritableStorageConnection(group_name);
具体方法是&＃xff1a;
protected boolean newWritableStorageConnection(String group_name) throws IOException, MyException{if (this.storageServer !&＃61; null){return false;}else{TrackerClient tracker &＃61; new TrackerClient();this.storageServer &＃61; tracker.getStoreStorage(this.trackerServer, group_name);if (this.storageServer &＃61;&＃61; null){throw new MyException("getStoreStorage fail, errno code: " &＃43; tracker.getErrorCode());}return true;}}
而这个方法里面的this.storageServer &＃61; tracker.getStoreStorage(this.trackerServer, group_name);
找进去发现每次trackerServer用完了&＃xff0c;就关闭连接了。
trackerServer.close();所以后面线程都被关闭了&＃xff0c;就报IO异常&＃xff0c;只要在方法③中new StorageClient前面再加一句&＃xff1a;TrackerServer trackerServer &＃61; trackerClient.getConnection();
就解决问题了。---------------------
作者&＃xff1a;S水流年
来源&＃xff1a;CSDN
原文&＃xff1a;https://blog.csdn.net/jianbo2233/article/details/79078500
版权声明&＃xff1a;本文为博主原创文章&＃xff0c;转载请附上博文链接&＃xff01;