添加上传文件接口，暂时还有问题

build.gradle

@@ -16,6 +16,7 @@ buildscript {
         caffeine_version = '2.6.1'
         shiro_version = '1.4.0'
         pinyin4j_version = '2.5.1'
+        rxJavaVersion='2.2.3'
     }
     repositories {
         mavenCentral()
@@ -102,6 +103,8 @@ subprojects {
         testCompile('org.springframework.boot:spring-boot-starter-test')
         // mq
         compile('com.fasterxml.jackson.module:jackson-module-kotlin:2.9.6')
+        compile("io.reactivex.rxjava2:rxjava:$rxJavaVersion")
+
 
         //只因用shiro核心包
         compile("org.apache.shiro:shiro-core:$shiro_version"){

cloud-bus/src/main/java/cn/gygxzc/tina/cloud/bus/EnableRabbitMessage.java

@@ -0,0 +1,21 @@
+package cn.gygxzc.tina.cloud.bus;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+/**
+ * Created by niantuo on 2018/10/27.
+ * 启用rabbitmq 服务线消息中间件
+ */
+
+@Configuration
+@Documented
+@Retention(java.lang.annotation.RetentionPolicy.RUNTIME)
+@Target({java.lang.annotation.ElementType.TYPE})
+@Import(RabbitConfiguration.class)
+public @interface EnableRabbitMessage {
+}

cloud-bus/src/main/java/cn/gygxzc/tina/cloud/bus/RabbitConfiguration.java

@@ -3,6 +3,7 @@ package cn.gygxzc.tina.cloud.bus;
 import cn.gygxzc.tina.cloud.bus.constant.QueueName;
 import org.springframework.amqp.core.Queue;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 
 /**
@@ -13,6 +14,7 @@ import org.springframework.context.annotation.Configuration;
  */
 
 @Configuration
+@ComponentScan
 public class RabbitConfiguration {
 
 

fastdfs-client/build.gradle

@@ -0,0 +1,15 @@
+group 'cn.gygxzc.cloud'
+version '1.0'
+
+sourceCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+    // https://mvnrepository.com/artifact/com.github.tobato/fastdfs-client
+    compile("com.github.tobato:fastdfs-client:1.26.3")
+
+}

fastdfs-client/src/main/java/cn/gygxzc/cloud/tina/fastdfs/client/EnableFastDFSClient.java

@@ -0,0 +1,23 @@
+package cn.gygxzc.cloud.tina.fastdfs.client;
+
+import com.github.tobato.fastdfs.FdfsClientConfig;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.EnableMBeanExport;
+import org.springframework.context.annotation.Import;
+import org.springframework.jmx.support.RegistrationPolicy;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+/**
+ * Created by niantuo on 2018/10/27.
+ * 开启dfs客户端支持
+ */
+@Configuration
+@Documented
+@Retention(java.lang.annotation.RetentionPolicy.RUNTIME)
+@Target({java.lang.annotation.ElementType.TYPE})
+@Import(value = {FastDFSClientConfiguration.class})
+public @interface EnableFastDFSClient {
+}

fastdfs-client/src/main/java/cn/gygxzc/cloud/tina/fastdfs/client/FastDFSClientConfiguration.java

@@ -0,0 +1,20 @@
+package cn.gygxzc.cloud.tina.fastdfs.client;
+
+import com.github.tobato.fastdfs.FdfsClientConfig;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.EnableMBeanExport;
+import org.springframework.context.annotation.Import;
+import org.springframework.jmx.support.RegistrationPolicy;
+
+/**
+ * Created by niantuo on 2018/10/27.
+ * 使用fastDFS 客户端配置
+ */
+
+@Configuration
+@ComponentScan
+@Import(FdfsClientConfig.class)
+@EnableMBeanExport(registration = RegistrationPolicy.IGNORE_EXISTING)
+public class FastDFSClientConfiguration {
+}

fastdfs-client/src/main/java/cn/gygxzc/cloud/tina/fastdfs/client/controller/FdfsDownloadController.java

@@ -0,0 +1,60 @@
+package cn.gygxzc.cloud.tina.fastdfs.client.controller;
+
+import com.github.tobato.fastdfs.proto.storage.DownloadCallback;
+import com.github.tobato.fastdfs.service.FastFileStorageClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * Created by niantuo on 2018/10/29.
+ * 文件下载或者说是
+ */
+@Controller
+@RequestMapping("/files")
+public class FdfsDownloadController {
+
+    @Autowired
+    private FastFileStorageClient storageClient;
+
+    @RequestMapping("/{group}/{path}")
+    public void download(@PathVariable("group") String group, @PathVariable("path") String path, HttpServletResponse response) {
+        InputStream inputStream = null;
+        OutputStream outputStream = null;
+        try {
+            inputStream = storageClient.downloadFile(group, path, ins -> ins);
+            outputStream = response.getOutputStream();
+            byte[] bytes = new byte[1024];
+            while ((inputStream.read(bytes)) != -1) {
+                outputStream.write(bytes, 0, bytes.length);
+            }
+            response.flushBuffer();
+        } catch (IOException e) {
+            throw new RuntimeException("文件下载失败，请重试。");
+        } finally {
+            if (inputStream != null) {
+                try {
+                    inputStream.close();
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }
+            }
+            if (outputStream != null) {
+                try {
+                    outputStream.flush();
+                    outputStream.close();
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }
+
+            }
+        }
+    }
+
+}

fastdfs-client/src/main/java/cn/gygxzc/cloud/tina/fastdfs/client/controller/FdfsUploadController.java

@@ -0,0 +1,118 @@
+package cn.gygxzc.cloud.tina.fastdfs.client.controller;
+
+import cn.gygxzc.cloud.tina.fastdfs.client.mata.CommonFileMata;
+import cn.gygxzc.cloud.tina.fastdfs.client.mata.FailedStorePath;
+import com.github.tobato.fastdfs.domain.MataData;
+import com.github.tobato.fastdfs.domain.StorePath;
+import com.github.tobato.fastdfs.service.FastFileStorageClient;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+import io.reactivex.SingleOnSubscribe;
+import io.reactivex.schedulers.Schedulers;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+
+/**
+ * Created by niantuo on 2018/10/27.
+ * 文件上传接口。
+ * 文件上传的接口
+ */
+@RestController
+@RequestMapping("/files/upload")
+public class FdfsUploadController {
+
+    @Autowired
+    private FastFileStorageClient storageClient;
+
+
+    /**
+     * 上传文件，需要自己附带一些信息
+     *
+     * @param file     文件本身信息
+     * @param fileMata 文件的自定义数据
+     * @return 文件的
+     */
+    @PostMapping
+    public Object upload(@RequestParam("file") MultipartFile file, CommonFileMata fileMata) {
+
+        Set<MataData> mataData = fileMata.mataData(file);
+        String fileExtName = file.getOriginalFilename();
+        try {
+            return storageClient.uploadFile(file.getInputStream(), file.getSize(), fileExtName, mataData);
+        } catch (IOException e) {
+            throw new RuntimeException("文件上传失败，请重试。");
+        }
+    }
+
+    /**
+     * 多文件上传。
+     * 暂时这种方式不知道会有什么风险。
+     * 把这个线程挂起来是不是也浪费了一些性能。
+     *
+     * @param files    文件信息
+     * @param fileMata 文件描述
+     * @return 文件的下载路径，不保证，顺序
+     */
+    @PostMapping("/multi")
+    public Object multiUpload(@RequestParam("files") List<MultipartFile> files, CommonFileMata fileMata) {
+
+        final Object mLock = new Object();
+        List<FailedStorePath> failedStorePaths = new ArrayList<>();
+        List<StorePath> storePaths = new ArrayList<>();
+        Observable.fromIterable(files)
+                .flatMapSingle(file -> uploadToFdfs(file, fileMata))
+                .doOnNext(storePath -> {
+                    if (storePath instanceof FailedStorePath) {
+                        failedStorePaths.add((FailedStorePath) storePath);
+                    } else {
+                        storePaths.add(storePath);
+                    }
+                })
+                .toList()
+                .subscribe(results -> {
+                    synchronized (mLock) {
+                        mLock.notifyAll();
+                    }
+                });
+        synchronized (mLock) {
+            try {
+                mLock.wait();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+        Map<String, Object> results = new HashMap<>();
+        results.put("paths", storePaths);
+        results.put("error", failedStorePaths);
+        return results;
+    }
+
+
+    /**
+     * 多文件上传，有可能只有一两个文件失败，这个时候，需要做一些处理.
+     * 设置文件上传最大的超时时长为两分钟，过长会影响用户体验
+     *
+     * @param file     上传的文件
+     * @param fileMata 文件的描述
+     * @return 文件的地址信息
+     */
+    private Single<StorePath> uploadToFdfs(MultipartFile file, CommonFileMata fileMata) {
+        return Single.create((SingleOnSubscribe<StorePath>) subscriber -> {
+            Set<MataData> mataData = fileMata.mataData(file);
+            String fileExtName = file.getOriginalFilename();
+            StorePath storePath = storageClient.uploadFile(file.getInputStream(), file.getSize(), fileExtName, mataData);
+            subscriber.onSuccess(storePath);
+        })
+                .timeout(2, TimeUnit.MINUTES)
+                .retry(3)
+                .onErrorReturn(throwable -> new FailedStorePath(file))
+                .subscribeOn(Schedulers.io());
+    }
+
+}

fastdfs-client/src/main/java/cn/gygxzc/cloud/tina/fastdfs/client/mata/CommonFileMata.java

@@ -0,0 +1,32 @@
+package cn.gygxzc.cloud.tina.fastdfs.client.mata;
+
+import com.github.tobato.fastdfs.domain.MataData;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Created by niantuo on 2018/10/27.
+ */
+
+public class CommonFileMata {
+    private String description;
+
+    public void setDescription(String description) {
+        this.description = description;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    public Set<MataData> mataData(MultipartFile file){
+        Set<MataData> mataDataSet = new HashSet<>();
+        mataDataSet.add(new MataData("originalName",file.getOriginalFilename()));
+        mataDataSet.add(new MataData("size",String.valueOf(file.getSize())));
+        mataDataSet.add(new MataData("name",file.getName()));
+        mataDataSet.add(new MataData("description",description));
+        return mataDataSet;
+    }
+}

fastdfs-client/src/main/java/cn/gygxzc/cloud/tina/fastdfs/client/mata/FailedStorePath.java

@@ -0,0 +1,42 @@
+package cn.gygxzc.cloud.tina.fastdfs.client.mata;
+
+import com.github.tobato.fastdfs.domain.StorePath;
+import org.springframework.web.multipart.MultipartFile;
+
+/**
+ * Created by niantuo on 2018/10/27.
+ * 上传失败的storePath
+ */
+
+public class FailedStorePath extends StorePath {
+
+    private String originalName;
+    private String fileName;
+
+
+    public FailedStorePath(MultipartFile file){
+        this.fileName = file.getName();
+        this.originalName = file.getOriginalFilename();
+    }
+
+    public FailedStorePath(String fileName, String originalName) {
+        this.originalName = originalName;
+        this.fileName = fileName;
+    }
+
+    public void setFileName(String fileName) {
+        this.fileName = fileName;
+    }
+
+    public void setOriginalName(String originalName) {
+        this.originalName = originalName;
+    }
+
+    public String getFileName() {
+        return fileName;
+    }
+
+    public String getOriginalName() {
+        return originalName;
+    }
+}

fastdfs-client/src/main/resources/application-fdfs.yml

@@ -0,0 +1,27 @@
+
+
+# fdfs 分布式文件服务器
+
+---
+spring:
+  profiles: dev
+fdfs:
+  connect-timeout: 600
+  so-timeout: 1500
+  thumb-image:
+    height: 150
+    width: 150
+  tracker-list:
+    - 192.168.1.206:22122
+---
+spring:
+  profiles: prod
+
+fdfs:
+  connect-timeout: 600
+  so-timeout: 1500
+  thumb-image:
+    height: 150
+    width: 150
+  tracker-list:
+  -  192.168.1.206:22122

fastdfs-client/src/test/java/cn/gygxzc/cloud/tina/fdfs/RxJavaTest.java

@@ -0,0 +1,60 @@
+package cn.gygxzc.cloud.tina.fdfs;
+
+import io.reactivex.Observable;
+import io.reactivex.Single;
+import io.reactivex.SingleEmitter;
+import io.reactivex.SingleOnSubscribe;
+import io.reactivex.schedulers.Schedulers;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Created by niantuo on 2018/10/27.
+ * 这种方式的确是可以使用的，但是不知道在实际过程中会不会表现很糟糕
+ */
+
+public class RxJavaTest {
+
+    public static void main(String[] args) {
+        List<Integer> integers = new ArrayList<>();
+        final Object mLock = new Object();
+        integers.add(1000);
+        integers.add(2000);
+        integers.add(2500);
+        integers.add(500);
+        integers.add(1500);
+        long startTime = System.currentTimeMillis();
+        Observable.fromIterable(integers)
+                .flatMapSingle(RxJavaTest::task)
+                .doOnNext(threadId->{
+                    System.out.printf("onNex: %d,threadId:%d \n",System.currentTimeMillis(),threadId);
+                })
+                .toList()
+                .subscribe(list->{
+                    System.out.printf("complete time: %d \n",System.currentTimeMillis());
+                    synchronized (mLock){
+                        mLock.notifyAll();
+                    }
+                });
+        synchronized (mLock){
+            try {
+                mLock.wait();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+        System.out.printf("main thread:%d  耗时：%d  \n",System.currentTimeMillis(),System.currentTimeMillis()-startTime);
+    }
+
+    private static Single<Long> task(long sleep){
+        return Single.create((SingleOnSubscribe<Long>) emitter -> {
+            System.out.printf("current thread: %d,start Time: %d \n",Thread.currentThread().getId(),System.currentTimeMillis());
+            Thread.sleep(sleep);
+            emitter.onSuccess(Thread.currentThread().getId());
+            System.out.printf("current thread: %d,end Time: %d \n",Thread.currentThread().getId(),System.currentTimeMillis());
+        }).subscribeOn(Schedulers.io())
+                .retry(3);
+    }
+
+}

settings.gradle

@@ -3,4 +3,5 @@ include 'zen-core'
 include 'zen-api'
 include 'zen-web'
 include 'cloud-bus'
+include 'fastdfs-client'
 

zen-web/build.gradle

@@ -13,6 +13,7 @@ buildscript {
 dependencies {
     compile project(":zen-api")
     compile project(':cloud-bus')
+    compile project(':fastdfs-client')
 }
 
 docker{

zen-web/src/main/kotlin/cn/gygxzc/envir/MainApplication.kt

@@ -1,5 +1,7 @@
 package cn.gygxzc.envir
 
+import cn.gygxzc.cloud.tina.fastdfs.client.EnableFastDFSClient
+import cn.gygxzc.tina.cloud.bus.EnableRabbitMessage
 import org.mybatis.spring.annotation.MapperScan
 import org.springframework.boot.SpringApplication
 import org.springframework.boot.autoconfigure.SpringBootApplication
@@ -13,7 +15,9 @@ import org.springframework.cloud.openfeign.EnableFeignClients
  */
 @EnableDiscoveryClient
 @EnableFeignClients
-@SpringBootApplication(scanBasePackages = ["cn.gygxzc", "com.gxzc"])
+@EnableFastDFSClient
+@EnableRabbitMessage
+@SpringBootApplication(scanBasePackages = ["cn.gygxzc.envir", "com.gxzc"])
 @MapperScan(basePackages = ["cn.gygxzc.**.dao", "cn.gygxzc.**.mapper"])
 class MainApplication : SpringBootServletInitializer() {
 

zen-web/src/main/resources/bootstrap.yml

@@ -7,7 +7,7 @@ spring:
     name: framework
   profiles:
     active: dev
-    include: orm,cache,msg
+    include: orm,cache,msg,fdfs
   cloud:
     config:
       profile: dev