java - 使用 Webflux 进行非阻塞 FileWalkTree 文件搜索

Question

我需要创建一个非阻塞功能，在其中搜索给定文件夹中的文本文件，并返回在其中找到的搜索词的计数。

我能够以阻塞方式执行测试。我想知道是否有人可以帮助我将其转换为非阻塞任务，以便每当文件扫描完成时都会传递结果，而无需等待所有文件都被扫描。

要点是:我不想等待所有文件都扫描完毕才开始将结果传递给客户端(Angular 应用程序)。

public interface SearchService {
    List<SearchResponse> search(SearchRequest searchRequest);
}

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.ToString;

import java.io.Serializable;

@ToString
@Getter
@RequiredArgsConstructor(staticName = "of")
public class SearchResponse implements Serializable {

    private final String server;
    private final String filePath;
    private final long count;
    private final boolean errorReadingFile;
}

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.ToString;

import javax.validation.constraints.NotNull;
import java.io.Serializable;


@ToString
@Getter
@RequiredArgsConstructor(staticName = "of")
public class SearchRequest implements Serializable {
    @NotNull
    private final String server;
    @NotNull
    private final String rootPath;
    @NotNull
    private final String searchTerm;
}

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.nio.file.*;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;


@Slf4j
@Service
public class FileSearchService implements SearchService {

    @Override
    public List<SearchResponse> search(SearchRequest searchRequest) {
        Path start = Paths.get(searchRequest.getRootPath());
        EnumSet<FileVisitOption> opts = EnumSet.of(FileVisitOption.FOLLOW_LINKS);
        int maxDepth = Integer.MAX_VALUE;
        SearchTermFileVisitor visitor = new SearchTermFileVisitor(searchRequest, new ArrayList<>());
        try {
            Files.walkFileTree(start,opts,maxDepth, visitor);
            return visitor.getSearchResponseList();
        } catch (IOException e) {
            System.out.println(e.getMessage());
        }
    }

}

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.IOException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Stream;

@Slf4j
@Getter
@RequiredArgsConstructor
public class SearchTermFileVisitor extends SimpleFileVisitor<Path> {
    private final SearchRequest searchRequest;
    private final List<SearchResponse> searchResponseList;

    private SearchResponse searchFileContent(Path path, SearchRequest searchRequest) {
        SearchResponse response;
        try (BufferedReader br = Files.newBufferedReader(path)) {
            response = SearchResponse.of(
                    searchRequest.getServer(),
                    Paths.get(path.toUri()).toString(),
                    countWordsInFile(searchRequest.getSearchTerm(), br.lines()),
                    false);
        } catch (Exception e) {
            response = SearchResponse.of(
                    searchRequest.getServer(),
                    path.toString(),
                    0,
                    true);
        }
        log.debug(response.toString());
        return response;
    }

    private int countWordsInFile(String searchTerm, Stream<String> linesStream) {
        return linesStream
                .parallel()
                .map(line -> countWordsInLine(line, searchTerm))
                .reduce(0, Integer::sum);
    }

    private int countWordsInLine(String line, String searchTerm) {
        Pattern pattern = Pattern.compile(searchTerm.toLowerCase());
        Matcher matcher = pattern.matcher(line.toLowerCase());

        int count = 0;
        int i = 0;
        while (matcher.find(i)) {
            count++;
            i = matcher.start() + 1;
        }
        return count;
    }

    private boolean isTextFile(Path path) throws IOException {
        String type = Files.probeContentType(path);
        if (type == null) {
            //type couldn't be determined, assume binary
            return false;
        } else if (type.startsWith("text")) {
            return true;
        } else {
            //type isn't text
            return false;
        }
    }

    @Override
    public FileVisitResult postVisitDirectory(Path dir, IOException exc) {
        log.debug("Visited: " + (Path) dir);
        return FileVisitResult.CONTINUE;
    }

    @Override
    public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
        return FileVisitResult.CONTINUE;
    }

    @Override
    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
        if (attrs.isRegularFile()
                && !attrs.isDirectory()
                && !attrs.isSymbolicLink()
                && isTextFile(file)) {
            searchResponseList.add(searchFileContent(file, searchRequest));
        }
        return FileVisitResult.CONTINUE;
    }

    @Override
    public FileVisitResult visitFileFailed(Path file, IOException exc) {
        return FileVisitResult.CONTINUE;
    }
}

测试用例:

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.net.URISyntaxException;
import java.nio.file.Paths;
import java.util.List;

class FileSearchServiceTest {

    private SearchService searchService = new FileSearchService();

    @Test
    void test_search_window_root_c_path() {
        SearchRequest sr = SearchRequest.of("localhost", "c:\\", "a");
        final List<SearchResponse> searchResult = searchService.search(sr);
        Assertions.assertNotNull(searchResult.size());
    }
}

我想使用WebFlux来一一接收结果，而不需要等待所有文件都扫描完成。

Answer 1

考虑 (1) 在单独的线程中执行搜索，(2) 使用观察者模式将中间结果接收回线程创建的代码(引用下文)，以及 (3) 将搜索线程连接到实现代码，以便完成后您可以返回结果列表。这意味着您需要将“this”传递给线程，以便它有一个引用来回调单独的方法以获得中间结果。下面的引用文献包含示例代码。

把它想象成一个 GUI。您可以在单独的线程中运行 GUI，并且对于每个按钮单击，它都会回调 Controller 代码(包括单击“退出”后的“完成”)。

回复:https://dzone.com/articles/the-observer-pattern-using-modern-java https://en.wikipedia.org/wiki/Observer_pattern