public BookieService(BookieConfiguration conf,
                         StatsLogger statsLogger)
            throws Exception {
        super(NAME, conf, statsLogger);
        this.server = new BookieServer(conf.getServerConf(), statsLogger);
    }
    
    
//BookieServer.java

	public BookieServer(ServerConfiguration conf, StatsLogger statsLogger)
            throws IOException, KeeperException, InterruptedException,
            BookieException, UnavailableException, CompatibilityException, SecurityException {
        this.conf = conf;
        //校验用户权限
        validateUser(conf);
        String configAsString;
        try {
            configAsString = conf.asJson();
            LOG.info(configAsString);
        } catch (ParseJsonException pe) {
            LOG.error("Got ParseJsonException while converting Config to JSONString", pe);
        }

        //构造内存分配器
        ByteBufAllocator allocator = getAllocator(conf);
        this.statsLogger = statsLogger;
        //构造NettyServer
        this.nettyServer = new BookieNettyServer(this.conf, null, allocator);
        try {
            //构造Bookie实例
            this.bookie = newBookie(conf, allocator);
        } catch (IOException | KeeperException | InterruptedException | BookieException e) {
            // interrupted on constructing a bookie
            this.nettyServer.shutdown();
            throw e;
        }
        final SecurityHandlerFactory shFactory;

        shFactory = SecurityProviderFactoryFactory
                .getSecurityProviderFactory(conf.getTLSProviderFactoryClass());
        //指定NettyServer的RequestProcessor
        this.requestProcessor = new BookieRequestProcessor(conf, bookie,
                statsLogger.scope(SERVER_SCOPE), shFactory, bookie.getAllocator());
        this.nettyServer.setRequestProcessor(this.requestProcessor);
    }

public static void main(String[] args) {
        int retCode = doMain(args);
        Runtime.getRuntime().exit(retCode);
    }

    static int doMain(String[] args) {

        ServerConfiguration conf;

        // 0. parse command line
        try {
            conf = parseCommandLine(args);
        } catch (IllegalArgumentException iae) {
            return ExitCode.INVALID_CONF;
        }

        // 1. building the component stack:
        LifecycleComponent server;
        try {
            server = buildBookieServer(new BookieConfiguration(conf));
        } catch (Exception e) {
            log.error("Failed to build bookie server", e);
            return ExitCode.SERVER_EXCEPTION;
        }

        // 2. start the server
        try {
            ComponentStarter.startComponent(server).get();
        } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
            // the server is interrupted
            log.info("Bookie server is interrupted. Exiting ...");
        } catch (ExecutionException ee) {
            log.error("Error in bookie shutdown", ee.getCause());
            return ExitCode.SERVER_EXCEPTION;
        }
        return ExitCode.OK;
    }

BasicParser parser = new BasicParser();
CommandLine cmdLine = parser.parse(BK_OPTS, args);

static {
        BK_OPTS.addOption("c", "conf", true, "Configuration for Bookie Server");
        BK_OPTS.addOption("withAutoRecovery", false,
                "Start Autorecovery service Bookie server");
        BK_OPTS.addOption("r", "readOnly", false,
                "Force Start a ReadOnly Bookie server");
        BK_OPTS.addOption("z", "zkserver", true, "Zookeeper Server");
        BK_OPTS.addOption("m", "zkledgerpath", true, "Zookeeper ledgers root path");
        BK_OPTS.addOption("p", "bookieport", true, "bookie port exported");
        BK_OPTS.addOption("j", "journal", true, "bookie journal directory");
        Option indexDirs = new Option ("i", "indexdirs", true, "bookie index directories");
        indexDirs.setArgs(10);
        BK_OPTS.addOption(indexDirs);
        Option ledgerDirs = new Option ("l", "ledgerdirs", true, "bookie ledgers directories");
        ledgerDirs.setArgs(10);
        BK_OPTS.addOption(ledgerDirs);
        BK_OPTS.addOption("h", "help", false, "Print help message");
    }

public static LifecycleComponentStack buildBookieServer(BookieConfiguration conf) throws Exception {
        LifecycleComponentStack.Builder serverBuilder = LifecycleComponentStack.newBuilder().withName("bookie-server");

        // 1. build stats provider
        StatsProviderService statsProviderService =
            new StatsProviderService(conf);
        StatsLogger rootStatsLogger = statsProviderService.getStatsProvider().getStatsLogger("");

        serverBuilder.addComponent(statsProviderService);
        log.info("Load lifecycle component : {}", StatsProviderService.class.getName());

        // 2. build bookie server
        BookieService bookieService =
            new BookieService(conf, rootStatsLogger);

        serverBuilder.addComponent(bookieService);
        log.info("Load lifecycle component : {}", BookieService.class.getName());

        if (conf.getServerConf().isLocalScrubEnabled()) {
            serverBuilder.addComponent(
                    new ScrubberService(
                            rootStatsLogger.scope(ScrubberStats.SCOPE),
                    conf, bookieService.getServer().getBookie().getLedgerStorage()));
        }

        // 3. build auto recovery
        if (conf.getServerConf().isAutoRecoveryDaemonEnabled()) {
            AutoRecoveryService autoRecoveryService =
                new AutoRecoveryService(conf, rootStatsLogger.scope(REPLICATION_SCOPE));

            serverBuilder.addComponent(autoRecoveryService);
            log.info("Load lifecycle component : {}", AutoRecoveryService.class.getName());
        }

        // 4. build http service
        if (conf.getServerConf().isHttpServerEnabled()) {
            BKHttpServiceProvider provider = new BKHttpServiceProvider.Builder()
                .setBookieServer(bookieService.getServer())
                .setServerConfiguration(conf.getServerConf())
                .setStatsProvider(statsProviderService.getStatsProvider())
                .build();
            HttpService httpService =
                new HttpService(provider, conf, rootStatsLogger);

            serverBuilder.addComponent(httpService);
            log.info("Load lifecycle component : {}", HttpService.class.getName());
        }

        // 5. build extra services
        String[] extraComponents = conf.getServerConf().getExtraServerComponents();
        if (null != extraComponents) {
            try {
                List<ServerLifecycleComponent> components = loadServerComponents(
                    extraComponents,
                    conf,
                    rootStatsLogger);
                for (ServerLifecycleComponent component : components) {
                    serverBuilder.addComponent(component);
                    log.info("Load lifecycle component : {}", component.getClass().getName());
                }
            } catch (Exception e) {
                if (conf.getServerConf().getIgnoreExtraServerComponentsStartupFailures()) {
                    log.info("Failed to load extra components '{}' - {}. Continuing without those components.",
                        StringUtils.join(extraComponents), e.getMessage());
                } else {
                    throw e;
                }
            }
        }

        return serverBuilder.build();
    }

@Override
    public void start() {
        components.forEach(component -> component.start());
    }

    @Override
    public void stop() {
        components.reverse().forEach(component -> component.stop());
    }

    @Override
    public void close() {
        components.reverse().forEach(component -> component.close());
    }

private static final String ZK_ADDR = "127.0.0.1:2181";

try {
    //初始化 BookKeeper Client 的方法
   
    BookKeeper bkClient = new BookKeeper(ZK_ADDR);           
       
} catch (Exception e) {
    e.printStackTrace();
    throw new RuntimeException(
            "There is an exception throw while creating the BookKeeper client.");
}

public class Mao {

    boolean flag;

    int a;
}

public class TestMao {

    public static void main(String[] args) {
        Mao mao = new Mao();
        System.out.println(ClassLayout.parseInstance(mao).toPrintable());
    }
}

INSERT INTO TEST(ID, NAME) VALUES(3000, 'aaa');

stat.execute("create table test(id int primary key, name varchar(255))");

//解析生成Command
CommandInterface command = conn.prepareCommand(sql, fetchSize);
boolean lazy = false;
boolean returnsResultSet;
synchronized (session) {
	setExecutingStatement(command);
	try {
	    if (command.isQuery()) {
	        returnsResultSet = true;
	        boolean scrollable = resultSetType != ResultSet.TYPE_FORWARD_ONLY;
	        boolean updatable = resultSetConcurrency == ResultSet.CONCUR_UPDATABLE;
	        ResultInterface result = command.executeQuery(maxRows, scrollable);
	        lazy = result.isLazy();
	        resultSet = new JdbcResultSet(conn, this, command, result, id,
	                closedByResultSet, scrollable, updatable);
	    } else {
	        returnsResultSet = false;
	        //执行Command命令
	        ResultWithGeneratedKeys result = command.executeUpdate(
	                conn.scopeGeneratedKeys() ? null : generatedKeysRequest);
	        updateCount = result.getUpdateCount();
	        ResultInterface gk = result.getGeneratedKeys();
	        if (gk != null) {
	            generatedKeys = new JdbcResultSet(conn, this, command, gk, id,
	                    false, true, false);
	        }
	    }
	} finally {
	    if (!lazy) {
	        setExecutingStatement(null);
	    }
	}
}

//获取schema 构建创建sql语句
Schema schema = getSchema();
CreateTable command = new CreateTable(session, schema);
command.setPersistIndexes(persistIndexes);
command.setTemporary(temp);
command.setGlobalTemporary(globalTemp);
command.setIfNotExists(ifNotExists);
command.setTableName(tableName);
command.setComment(readCommentIf());

Column column = parseColumnForTable(columnName, true, true);
if (column.isAutoIncrement() && column.isPrimaryKey()) {
    column.setPrimaryKey(false);
    IndexColumn[] cols = { new IndexColumn() };
    cols[0].columnName = column.getName();
    AlterTableAddConstraint pk = new AlterTableAddConstraint(
            session, schema, false);
    pk.setType(CommandInterface.ALTER_TABLE_ADD_CONSTRAINT_PRIMARY_KEY);
    pk.setTableName(tableName);
    pk.setIndexColumns(cols);
    command.addConstraintCommand(pk);
}
command.addColumn(column);

changePrimaryKeysToNotNull(data.columns);
data.id = getObjectId();
data.create = create;
data.session = session;
//根据schema创建table
Table table = getSchema().createTable(data);

ArrayList<Sequence> sequences = generateSequences(data.columns, data.temporary);
table.setComment(comment);
if (isSessionTemporary) {
    if (onCommitDrop) {
        table.setOnCommitDrop(true);
    }
    if (onCommitTruncate) {
        table.setOnCommitTruncate(true);
    }
    session.addLocalTempTable(table);
} else {
    db.lockMeta(session);
    db.addSchemaObject(session, table);
}

jdbc:h2:~/test;MV_STORE=false

//DataBase.java
/**
    * 初始化存储引擎
    *
    * @return
    */
   public PageStore getPageStore() {
       //MV_STORE=false/true 默认true 表明存储引擎使用 MvStore
       if (dbSettings.mvStore) {
           if (store == null) {
               //传dataBase对象进去 初始化 MVTableEngine.Store(包含 MvStore TransactionStore)
               store = MVTableEngine.init(this);
           }
           return null;
       }
       //存储引擎使用 pageStore
       synchronized (this) {
           if (pageStore == null) {
               pageStore = new PageStore(this, databaseName +
                       Constants.SUFFIX_PAGE_FILE, accessModeData, cacheSize);
               if (pageSize != Constants.DEFAULT_PAGE_SIZE) {
                   pageStore.setPageSize(pageSize);
               }
               if (!readOnly && fileLockMethod == FileLockMethod.FS) {
                   pageStore.setLockFile(true);
               }
               pageStore.setLogMode(logMode);
               pageStore.open();
           }
           return pageStore;
       }
   }

public static Store init(final Database db) {
        Store store = db.getStore();
        if (store != null) {
            return store;
        }
        byte[] key = db.getFileEncryptionKey();
        String dbPath = db.getDatabasePath();
        MVStore.Builder builder = new MVStore.Builder();
        store = new Store();
        boolean encrypted = false;
        if (dbPath != null) {
            String fileName = dbPath + Constants.SUFFIX_MV_FILE;
            MVStoreTool.compactCleanUp(fileName);
            builder.fileName(fileName);
            builder.pageSplitSize(db.getPageSize());
            if (db.isReadOnly()) {
                builder.readOnly();
            } else {
                // possibly create the directory
                boolean exists = FileUtils.exists(fileName);
                if (exists && !FileUtils.canWrite(fileName)) {
                    // read only
                } else {
                    String dir = FileUtils.getParent(fileName);
                    FileUtils.createDirectories(dir);
                }
                int autoCompactFillRate = db.getSettings().maxCompactCount;
                if (autoCompactFillRate <= 100) {
                    builder.autoCompactFillRate(autoCompactFillRate);
                }
            }
            if (key != null) {
                encrypted = true;
                builder.encryptionKey(decodePassword(key));
            }
            if (db.getSettings().compressData) {
                builder.compress();
                // use a larger page split size to improve the compression ratio
                builder.pageSplitSize(64 * 1024);
            }
            builder.backgroundExceptionHandler(new UncaughtExceptionHandler() {

                @Override
                public void uncaughtException(Thread t, Throwable e) {
                    db.setBackgroundException(DbException.convert(e));
                }

            });
            // always start without background thread first, and if necessary,
            // it will be set up later, after db has been fully started,
            // otherwise background thread would compete for store lock
            // with maps opening procedure
            builder.autoCommitDisabled();
        }
        store.open(db, builder, encrypted);
        db.setStore(store);
        return store;
    }
    
    
    /**
         * Open the store for this database.
         *
         * @param db the database
         * @param builder the builder
         * @param encrypted whether the store is encrypted
         */
        void open(Database db, MVStore.Builder builder, boolean encrypted) {
            this.encrypted = encrypted;
            try {
                //初始化MvStore
                this.mvStore = builder.open();
                FileStore fs = mvStore.getFileStore();
                if (fs != null) {
                    this.fileName = fs.getFileName();
                }
                if (!db.getSettings().reuseSpace) {
                    mvStore.setReuseSpace(false);
                }
                mvStore.setVersionsToKeep(0);
                //初始化 TransactionStore
                this.transactionStore = new TransactionStore(mvStore,
                        new ValueDataType(db, null), db.getLockTimeout());
            } catch (IllegalStateException e) {
                throw convertIllegalStateException(e);
            }
        }

MVStore(Map<String, Object> config) {
        recoveryMode = config.containsKey("recoveryMode");
        compressionLevel = DataUtils.getConfigParam(config, "compress", 0);
        String fileName = (String) config.get("fileName");
        FileStore fileStore = (FileStore) config.get("fileStore");
        fileStoreIsProvided = fileStore != null;
        if(fileStore == null && fileName != null) {
            fileStore = new FileStore();
        }
        this.fileStore = fileStore;

        int pgSplitSize = 48; // for "mem:" case it is # of keys
        CacheLongKeyLIRS.Config cc = null;
        if (this.fileStore != null) {
            int mb = DataUtils.getConfigParam(config, "cacheSize", 16);
            if (mb > 0) {
                cc = new CacheLongKeyLIRS.Config();
                cc.maxMemory = mb * 1024L * 1024L;
                Object o = config.get("cacheConcurrency");
                if (o != null) {
                    cc.segmentCount = (Integer)o;
                }
            }
            pgSplitSize = 16 * 1024;
        }
        if (cc != null) {
            cache = new CacheLongKeyLIRS<>(cc);
        } else {
            cache = null;
        }

        pgSplitSize = DataUtils.getConfigParam(config, "pageSplitSize", pgSplitSize);
        // Make sure pages will fit into cache
        if (cache != null && pgSplitSize > cache.getMaxItemSize()) {
            pgSplitSize = (int)cache.getMaxItemSize();
        }
        pageSplitSize = pgSplitSize;
        keysPerPage = DataUtils.getConfigParam(config, "keysPerPage", 48);
        backgroundExceptionHandler =
                (UncaughtExceptionHandler)config.get("backgroundExceptionHandler");
        meta = new MVMap<>(this);
        if (this.fileStore != null) {
            retentionTime = this.fileStore.getDefaultRetentionTime();
            // 19 KB memory is about 1 KB storage
            int kb = Math.max(1, Math.min(19, Utils.scaleForAvailableMemory(64))) * 1024;
            kb = DataUtils.getConfigParam(config, "autoCommitBufferSize", kb);
            autoCommitMemory = kb * 1024;
            autoCompactFillRate = DataUtils.getConfigParam(config, "autoCompactFillRate", 90);
            char[] encryptionKey = (char[]) config.get("encryptionKey");
            try {
                if (!fileStoreIsProvided) {
                    boolean readOnly = config.containsKey("readOnly");
                    //创建数据库文件 *.db
                    this.fileStore.open(fileName, readOnly, encryptionKey);
                }
                //往 *.db 写数据
                if (this.fileStore.size() == 0) {
                    creationTime = getTimeAbsolute();
                    lastCommitTime = creationTime;
                    storeHeader.put(HDR_H, 2);
                    storeHeader.put(HDR_BLOCK_SIZE, BLOCK_SIZE);
                    storeHeader.put(HDR_FORMAT, FORMAT_WRITE);
                    storeHeader.put(HDR_CREATED, creationTime);
                    writeStoreHeader();
                } else {
                    // there is no need to lock store here, since it is not opened yet,
                    // just to make some assertions happy, when they ensure single-threaded access
                    storeLock.lock();
                    try {
                        readStoreHeader();
                    } finally {
                        storeLock.unlock();
                    }
                }
            } catch (IllegalStateException e) {
                panic(e);
            } finally {
                if (encryptionKey != null) {
                    Arrays.fill(encryptionKey, (char) 0);
                }
            }
            lastCommitTime = getTimeSinceCreation();

            scrubMetaMap();

            // setAutoCommitDelay starts the thread, but only if
            // the parameter is different from the old value
            int delay = DataUtils.getConfigParam(config, "autoCommitDelay", 1000);
            setAutoCommitDelay(delay);
        } else {
            autoCommitMemory = 0;
            autoCompactFillRate = 0;
        }
    }