indexmerger.java

来自「jsr170接口的java实现。是个apache的开源项目。」· Java 代码 · 共 500 行 · 第 1/2 页

    }

    /**
     * Implements the index merging.
     */
    public void run() {
        for (;;) {
            boolean isIdle = false;
            if (mergeTasks.size() == 0) {
                mergerIdle.release();
                isIdle = true;
            }
            Merge task = (Merge) mergeTasks.remove();
            if (task == QUIT) {
                mergerIdle.release();
                break;
            }
            if (isIdle) {
                try {
                    mergerIdle.acquire();
                } catch (InterruptedException e) {
                    Thread.interrupted();
                    log.warn("Unable to acquire mergerIdle sync");
                }
            }

            log.debug("accepted merge request");

            // reset deleted documents
            deletedDocuments.clear();

            // get readers
            String[] names = new String[task.indexes.length];
            for (int i = 0; i < task.indexes.length; i++) {
                names[i] = task.indexes[i].name;
            }
            try {
                log.debug("create new index");
                PersistentIndex index = multiIndex.getOrCreateIndex(null, true);
                boolean success = false;
                try {

                    log.debug("get index readers from MultiIndex");
                    IndexReader[] readers = multiIndex.getIndexReaders(names, this);
                    try {
                        // do the merge
                        long time = System.currentTimeMillis();
                        index.addIndexes(readers);
                        time = System.currentTimeMillis() - time;
                        int docCount = 0;
                        for (int i = 0; i < readers.length; i++) {
                            docCount += readers[i].numDocs();
                        }
                        log.info("merged " + docCount + " documents in " + time + " ms into " + index.getName() + ".");
                    } finally {
                        for (int i = 0; i < readers.length; i++) {
                            try {
                                readers[i].close();
                            } catch (IOException e) {
                                log.warn("Unable to close IndexReader: " + e);
                            }
                        }
                    }

                    // inform multi index
                    // if we cannot get the sync immediately we have to quit
                    if (!indexReplacement.attempt(0)) {
                        log.debug("index merging canceled");
                        break;
                    }
                    try {
                        log.debug("replace indexes");
                        multiIndex.replaceIndexes(names, index, deletedDocuments);
                    } finally {
                        indexReplacement.release();
                    }

                    success = true;

                } finally {
                    if (!success) {
                        // delete index
                        log.debug("deleting index " + index.getName());
                        multiIndex.deleteIndex(index);
                    }
                }
            } catch (Throwable e) {
                log.error("Error while merging indexes: " + e);
            }
        }
        log.info("IndexMerger terminated");
    }

    //-----------------------< merge properties >-------------------------------

    /**
     * The merge factor.
     */
    public void setMergeFactor(int mergeFactor) {
        this.mergeFactor = mergeFactor;
    }


    /**
     * The initial threshold for number of documents to merge to a new index.
     */
    public void setMinMergeDocs(int minMergeDocs) {
        this.minMergeDocs = minMergeDocs;
    }

    /**
     * The maximum number of document to merge.
     */
    public void setMaxMergeDocs(int maxMergeDocs) {
        this.maxMergeDocs = maxMergeDocs;
    }

    //------------------------------< internal >--------------------------------

    /**
     * Implements a simple struct that holds the name of an index and how
     * many document it contains. <code>Index</code> is comparable using the
     * number of documents it contains.
     */
    private static final class Index implements Comparable {

        /**
         * The name of the index.
         */
        private final String name;

        /**
         * The number of documents the index contains.
         */
        private final int numDocs;

        /**
         * Creates a new index struct.
         *
         * @param name name of an index.
         * @param numDocs number of documents it contains.
         */
        Index(String name, int numDocs) {
            this.name = name;
            this.numDocs = numDocs;
        }

        /**
         * Indexes are first ordered by {@link #numDocs} and then by {@link
         * #name}.
         *
         * @param o the other <code>Index</code>.
         * @return a negative integer, zero, or a positive integer as this
         *         Index is less than, equal to, or greater than the specified
         *         Index.
         */
        public int compareTo(Object o) {
            Index other = (Index) o;
            int val = numDocs < other.numDocs ? -1 : (numDocs == other.numDocs ? 0 : 1);
            if (val != 0) {
                return val;
            } else {
                return name.compareTo(other.name);
            }
        }

        /**
         * @inheritDoc
         */
        public String toString() {
            return name + ":" + numDocs;
        }
    }

    /**
     * Defines a merge task, to merge a couple of indexes into a new index.
     */
    private static final class Merge {

        private final Index[] indexes;

        /**
         * Merge task, to merge <code>indexes</code> into a new index with
         * <code>name</code>.
         *
         * @param indexes the indexes to merge.
         */
        Merge(Index[] indexes) {
            this.indexes = new Index[indexes.length];
            System.arraycopy(indexes, 0, this.indexes, 0, indexes.length);
        }
    }

    /**
     * Implements a <code>List</code> with a document limit value. An
     * <code>IndexBucket</code> contains {@link Index}es with documents less
     * or equal the document limit of the bucket.
     */
    private static final class IndexBucket extends ArrayList {

        /**
         * The lower document limit.
         */
        private final int lower;

        /**
         * The upper document limit.
         */
        private final int upper;

        /**
         * Flag indicating if indexes in this bucket can be merged.
         */
        private final boolean allowMerge;

        /**
         * Creates a new <code>IndexBucket</code>. Limits are both inclusive.
         *
         * @param lower document limit.
         * @param upper document limit.
         * @param allowMerge if indexes in this bucket can be merged.
         */
        IndexBucket(int lower, int upper, boolean allowMerge) {
            this.lower = lower;
            this.upper = upper;
            this.allowMerge = allowMerge;
        }

        /**
         * Returns <code>true</code> if the number of documents fit in this
         * <code>IndexBucket</code>; otherwise <code>false</code>
         *
         * @param numDocs the number of documents.
         * @return <code>true</code> if <code>numDocs</code> fit.
         */
        boolean fits(int numDocs) {
            return numDocs >= lower && numDocs <= upper;
        }

        /**
         * Returns <code>true</code> if indexes in this bucket can be merged.
         *
         * @return <code>true</code> if indexes in this bucket can be merged.
         */
        boolean allowsMerge() {
            return allowMerge;
        }
    }
}