By IceMimosa

更新日期:

文章目录
  1. BookKeeper基本使用
  2. 1. 安装
  3. 2. 基本的代码操作
    1. 2.1 使用 ledger API
    2. 2.2 使用 Ledger Advanced API
    3. 2.3 使用 DistributedLog

[TOC]

BookKeeper基本使用

在看pulsar的时候，发现它底层的存储使用的是Apache BookKeeper，所以对他的使用做下基本介绍。在看Bookkeeper的时候，发现有几个挺不错的文章，链接如下：

下面做下基本API的使用，想要深入可以参考源码或者文章。

1. 安装

注意: BookKeeper依赖zookeeper集群，请先提前准备

👉 点击连接下载BookKeeper最新版本

解压文件到相应目录，目录的基本结构如下，你也可以clone代码自己编译打包：

1
2
3
4
5
6
7
8
9
bookkeeper-server-4.9.1
	-- LICENSE
	-- NOTICE
	-- README.md
	-- bin        // 启动脚本相关
	-- conf       // 配置文件
	-- deps
	-- lib
	-- logs
  • 修改配置文件 conf/bk_server.conf
配置 说明 默认值 备注
bookiePort bookie实例监听的TCP端口 3181
advertisedAddress 绑定的主机名 可以改成 localhost
zkServers 逗号分隔的zk集群地址 localhost:2181
journalDirectories bookie存放write-ahead log (WAL)的目录，逗号分隔 /tmp/bk-txn 可以修改一下, 如/usr/local/var/bookkeeper/bk-txn
ledgerDirectories bookie存放数据ledger的目录，逗号分隔 /tmp/bk-data 可以修改一下, 如/usr/local/var/bookkeeper/bk-data
httpServerEnabled 是否打开http服务 false 可以改为true，api地址
httpServerPort http服务端口 8080 可以改为 3182
dlog.bkcEnsembleSize 默认bookie的最小节点数量 3 本地可换成1
dlog.bkcWriteQuorumSize 默认日志的冗余数量 2
dlog.bkcAckQuorumSize 日志写入成功多少份才会向client返回成功 2
  • 初始化元数据（初始化一次即可）
1
2
3
$ bin/bookkeeper shell metaformat

# 如果需要删除zk中的元数据：rmr /ledgers
  • 启动bookie服务
1
2
3
4
5
# 后台进程启动
$ bin/bookkeeper-daemon.sh start bookie

# 或者前台进程启动：
# $ bin/bookkeeper bookie

注：你也可以使用local的模式启动，参考文档

2. 基本的代码操作

BookKeeper提供的多套API来操作：

如果你想要使用起来像日志一样滚动、按需保留数据等操作的话，直接使用Ledger API可能需要你手动管理Ledgers列表，比较麻烦。所以在这种场景下建议使用DistributedLog API。

2.1 使用 ledger API

  • 加入maven依赖
1
2
3
4
5
6
7
8
9
<!-- in your <properties> block -->
<bookkeeper.version>4.9.1</bookkeeper.version>

<!-- in your <dependencies> block -->
<dependency>
  <groupId>org.apache.bookkeeper</groupId>
  <artifactId>bookkeeper-server</artifactId>
  <version>${bookkeeper.version}</version>
</dependency>
  • 初始化客户端（语言选择kotlin）
1
2
3
4
5
6
val bookKeeper = run {
    val conf = ClientConfiguration()
    conf.zkServers = "127.0.0.1:2181"
    conf.addEntryTimeout = 2000
    BookKeeper.forConfig(conf).build()
}
  • 创建 Ledger
1
2
3
4
5
6
7
8
9
10
// 也可以使用 asyncCreateLedger 异步创建
// 我本地只有一个 bookie, 所以传入1和1。默认是3和2
val ledger = bookKeeper.createLedger(1, 1, BookKeeper.DigestType.MAC, "my-password".toByteArray())
println(ledger.id)
// 输出: 0

// 写入数据 entry
ledger.addEntry("Hello Bookkeeper".toByteArray())

ledger.close()
  • 读取数据
1
2
3
4
5
6
7
// 0 表示 ledgerID
val ledger = bookKeeper.openLedger(0L, BookKeeper.DigestType.MAC, "my-password".toByteArray())
// 读取 entry
val entry = ledger.readLastEntry()
println(entry.entry.toString(Charset.defaultCharset()))

ledger.close()

其他操作见：ledger API

2.2 使用 Ledger Advanced API

从 v4.5.0 版本开始，增加 Ledger Advanced API，使用方式基本和Ledger API差不多（在API尾部加了个Adv，如创建变成了createLedgerAdv），功能增加了ledgerID的自定义传入等。

2.3 使用 DistributedLog

参考：https://bookkeeper.apache.org/distributedlog/docs/latest/user_guide/api/core

pulsar
pulsar
文章目录
  1. BookKeeper基本使用
  2. 1. 安装
  3. 2. 基本的代码操作
    1. 2.1 使用 ledger API
    2. 2.2 使用 Ledger Advanced API
    3. 2.3 使用 DistributedLog
∧