<commit_before>package com.virjar.vscrawler.core.seed;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.math.NumberUtils;

import com.google.common.collect.Maps;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnel;
import com.google.common.hash.PrimitiveSink;
import com.sleepycat.je.*;
import com.virjar.vscrawler.core.event.support.AutoEventRegistry;
import com.virjar.vscrawler.core.event.systemevent.CrawlerConfigChangeEvent;
import com.virjar.vscrawler.core.event.systemevent.CrawlerEndEvent;
import com.virjar.vscrawler.core.event.systemevent.FirstSeedPushEvent;
import com.virjar.vscrawler.core.event.systemevent.NewSeedArrivalEvent;
import com.virjar.vscrawler.core.util.PathResolver;
import com.virjar.vscrawler.core.util.SingtonObjectHolder;
import com.virjar.vscrawler.core.util.VSCrawlerCommonUtil;
import com.virjar.vscrawler.core.util.VSCrawlerConstant;

import lombok.extern.slf4j.Slf4j;

/**
 * Created by virjar on 17/5/14. <br/>
 * 伯克利数据库,用来管理种子任务
 *
 * @author virjar
 * @since 0.0.1
 */
@Slf4j
public class BerkeleyDBSeedManager implements CrawlerConfigChangeEvent, NewSeedArrivalEvent, CrawlerEndEvent {

    private Environment env;

    private String dbFilePath;

    private InitSeedSource initSeedSource;

    private SeedKeyResolver seedKeyResolver;

    private BloomFilter<Seed> bloomFilter;

    private DatabaseConfig databaseConfig;

    private DatabaseEntry iteratorKey = new DatabaseEntry();
    private DatabaseEntry iteratorValue = new DatabaseEntry();

    private AtomicBoolean isSeedEmpty = new AtomicBoolean(false);

    // 内部cache,数据加载到内存,以对象的方式存储,避免db操作带来锁的压力
    private ConcurrentLinkedQueue<Seed> ramCache = new ConcurrentLinkedQueue<>();
    // 所有正在处理的种子
    private Map<String, Seed> runningSeeds = Maps.newConcurrentMap();
    private volatile boolean isClosed = false;
    private int cacheSize;

    /**
     * 这个方法和pool必须在同一个线程里面
     */
    public void init() {
        // 移植游标
        archive();
    }

    public BerkeleyDBSeedManager(InitSeedSource initSeedSource, SeedKeyResolver seedKeyResolver, int cacheSize) {
        this.initSeedSource = initSeedSource;
        this.seedKeyResolver = seedKeyResolver;
        this.cacheSize = cacheSize;
        // 配置数据库环境
        configEnv();

        // 布隆过滤器数据还原
        buildBloomFilterInfo();

        // 移植初始种子信息
        migrateInitSeed();

        // 监听消息
        AutoEventRegistry.getInstance().registerObserver(this);
    }

    private synchronized void loadCache() {
        Database iteratorDatabases = env.openDatabase(null, "crawlSeed", databaseConfig);
        Cursor cursor = iteratorDatabases.openCursor(null, CursorConfig.DEFAULT);
        Database finishedSeedDatabase = env.openDatabase(null, "finishedSeed", databaseConfig);
        try {
            while (cursor.getNext(iteratorKey, iteratorValue, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
                Seed ret = VSCrawlerCommonUtil.transferStringToSeed(new String(iteratorValue.getData()));
                cursor.delete();// 删除当前数据
                if (!ret.needEnd()) {
                    ramCache.offer(ret);
                    if (ramCache.size() >= cacheSize) {
                        break;<commit_msg>udapte<commit_after>package com.virjar.vscrawler.core.seed;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.math.NumberUtils;

import com.google.common.collect.Maps;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnel;
import com.google.common.hash.PrimitiveSink;
import com.sleepycat.je.*;
import com.virjar.vscrawler.core.event.support.AutoEventRegistry;
import com.virjar.vscrawler.core.event.systemevent.CrawlerConfigChangeEvent;
import com.virjar.vscrawler.core.event.systemevent.CrawlerEndEvent;
import com.virjar.vscrawler.core.event.systemevent.FirstSeedPushEvent;
import com.virjar.vscrawler.core.event.systemevent.NewSeedArrivalEvent;
import com.virjar.vscrawler.core.util.PathResolver;
import com.virjar.vscrawler.core.util.SingtonObjectHolder;
import com.virjar.vscrawler.core.util.VSCrawlerCommonUtil;
import com.virjar.vscrawler.core.util.VSCrawlerConstant;

import lombok.extern.slf4j.Slf4j;

/**
 * Created by virjar on 17/5/14. <br/>
 * 伯克利数据库,用来管理种子任务
 *
 * @author virjar
 * @since 0.0.1
 */
@Slf4j
public class BerkeleyDBSeedManager implements CrawlerConfigChangeEvent, NewSeedArrivalEvent, CrawlerEndEvent {

    private Environment env;

    private String dbFilePath;

    private InitSeedSource initSeedSource;

    private SeedKeyResolver seedKeyResolver;

    private BloomFilter<Seed> bloomFilter;

    private DatabaseConfig databaseConfig;

    private DatabaseEntry iteratorKey = new DatabaseEntry();
    private DatabaseEntry iteratorValue = new DatabaseEntry();

    private AtomicBoolean isSeedEmpty = new AtomicBoolean(false);

    // 内部cache,数据加载到内存,以对象的方式存储,避免db操作带来锁的压力
    private ConcurrentLinkedQueue<Seed> ramCache = new ConcurrentLinkedQueue<>();
    // 所有正在处理的种子
    private Map<String, Seed> runningSeeds = Maps.newConcurrentMap();
    private volatile boolean isClosed = false;
    private int cacheSize;

    /**
     * 这个方法和pool必须在同一个线程里面
     */
    public void init() {
        // 移植游标
        //archive(); //大量数据会导致程序很慢,而且似乎没有意义
    }

    public BerkeleyDBSeedManager(InitSeedSource initSeedSource, SeedKeyResolver seedKeyResolver, int cacheSize) {
        this.initSeedSource = initSeedSource;
        this.seedKeyResolver = seedKeyResolver;
        this.cacheSize = cacheSize;
        // 配置数据库环境
        configEnv();

        // 布隆过滤器数据还原
        buildBloomFilterInfo();

        // 移植初始种子信息
        migrateInitSeed();

        // 监听消息
        AutoEventRegistry.getInstance().registerObserver(this);
    }

    private synchronized void loadCache() {
        Database iteratorDatabases = env.openDatabase(null, "crawlSeed", databaseConfig);
        Cursor cursor = iteratorDatabases.openCursor(null, CursorConfig.DEFAULT);
        Database finishedSeedDatabase = env.openDatabase(null, "finishedSeed", databaseConfig);
        try {
            while (cursor.getNext(iteratorKey, iteratorValue, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
                Seed ret = VSCrawlerCommonUtil.transferStringToSeed(new String(iteratorValue.getData()));
                cursor.delete();// 删除当前数据
                if (!ret.needEnd()) {
                    ramCache.offer(ret);
                    if (ramCache.size() >= cacheSize) {
                        break;
{'fraction_non_alphanumeric': 0.09650837988826816, 'fraction_numerical': 0.0033519553072625698, 'mean_word_length': 3.65, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6893485', 'n_tokens_mistral': 2326, 'n_tokens_neox': 2315, 'n_words': 384}