导读:IndexedDB 是一个 NOSQL 数据库,可以异步操作,支持事务,可存储 JSON 数据并且用索引迭代,兼容性好,适用于做大量的数据存储,本文主要探讨直接使用 IndexedDB 来存储数据时的一些情况以及遇到的问题。

当一个 Javascript 程序需要在浏览器端存储数据时,你有以下几个选择:
很明显,只有 IndexedDB 适用于做大量的数据存储。但是直接使用 IndexedDB 也会碰到几个问题:
本篇文章将结合笔者的实践经验,就以上问题来进行相关探索。

有这样一个场景,客户端产生大量的日志并存放若干日志。在发生某些错误时(或者长连接得到服务器的指令时)可拉取本地全部日志内容并发请求上报。
如图所示:

这是一个很好的设计到了 IndexedDB CRUD 场景的操作,在这里,我们只关注 IndexedDB 存储这部分。有关于 IndexedDB 的基础概念,如仓库 IDBObjectStore、索引 IDBIndex、游标 IDBCursor、事务 IDBTransaction,

我们知道 IndexedDB 是事务驱动的,打开一个数据库 db_test,创建 store log,并以 time 为索引。
- class Database {
- constructor(options = {}) {
- if (typeof indexedDB === 'undefined') {
- throw new Error('indexedDB is unsupported!')
- return
- }
- this.name = options.name
- this.db = null
- this.version = options.version || 1
- }
-
-
- createDB () {
- return new Promise((resolve, reject) => {
- // 为了本地调试,数据库先删除后建立
- indexedDB.deleteDatabase(this.name);
- const request = indexedDB.open(this.name);
- // 当数据库升级时,触发 onupgradeneeded 事件。
- // 升级是指该数据库首次被创建,或调用 open() 方法时指定的数据库的版本号高于本地已有的版本。
- request.onupgradeneeded = () => {
- const db = request.result;
- window.db = db
- console.log('db onupgradeneeded')
- // 在这里创建 store
- this.createStore(db)
- };
-
-
- // 打开成功的回调函数
- request.onsuccess = () => {
- resolve(request.result)
- this.db = request.result
- };
- // 打开失败的回调函数
- request.onerror = function(event) {
- reject(event)
- }
- })
- }
-
-
- createStore(db) {
- if (!db.objectStoreNames.contains('log')) {
- // 创建表
- const objectStore = db.createObjectStore('log', {
- keyPath: 'id',
- autoIncrement: true
- });
- // time 为索引
- objectStore.createIndex('time', 'time');
- }
- }
- }
调用语句如下:
- (async function() {
- const database = new Database({ name: 'db_test' })
- await database.createDB()
- console.log(database)
- // Database {name: 'db_test', db: IDBDatabase, version: 1}
- // db: IDBDatabase
- // name: "db_test"
- // objectStoreNames: DOMStringList {0: 'log', length: 1}
- // onabort: null
- // onclose: null
- // onerror: null
- // onversionchange: null
- // version: 1
- // [[Prototype]]: IDBDatabase
- // name: "db_test"
- // version: 1
- // [[Prototype]]: Object
- })()

增删改操作
当日志插入一条数据,我们需要提交一个事务,事务里对 store 进行 add 操作。
- const db = window.db;
- const transaction = db.transaction('log', 'readwrite')
- const store = transaction.objectStore('log')
-
-
- const storeRequest = store.add(data);
-
-
- storeRequest.onsuccess = function(event) {
- console.log('add onsuccess, affect rows ', event.target.result);
- resolve(event.target.result)
- };
-
-
- storeRequest.onerror = function(event) {
- reject(event);
- };
由于每次的增删改查都需要打开一个 transaction,这样的调用不免显得繁琐,我们需要一些步骤来简化,提供 ES6 promise 形式的 API。
- class Database {
- // ... 省略打开数据库的过程
- // constructor(options = {}) {}
- // createDB() {}
- // createStore() {}
- add (data) {
- return new Promise((resolve, reject) => {
- const db = this.db;
- const transaction = db.transaction('log', 'readwrite')
- const store = transaction.objectStore('log')
- const request = store.add(data);
-
-
- request.onsuccess = event => resolve(event.target.result);
- request.onerror = event => reject(event);
- })
- }
- put (data) {
- return new Promise((resolve, reject) => {
- const db = this.db;
- const transaction = db.transaction('log', 'readwrite')
- const store = transaction.objectStore('log')
- const request = store.put(data);
-
-
- request.onsuccess = event => resolve(event.target.result);
- request.onerror = event => reject(event);
- })
- }
- // delete
- delete (id) {
- return new Promise((resolve, reject) => {
- const db = this.db;
- const transaction = db.transaction('log', 'readwrite')
- const store = transaction.objectStore('log')
- const request = store.delete(id)
-
-
- request.onsuccess = event => resolve(event.target.result);
- request.onerror = event => reject(event);
- })
- }
- }
调用代码如下:
- (async function() {
- const db = new Database({ name: 'db_test' })
- await db.createDB()
-
- const row1 = await db.add({time: new Date().getTime(), body: 'log 1' })
- // {id: 1, time: new Date().getTime(), body: 'log 2' }
-
-
- await db.add({time: new Date().getTime(), body: 'log 2' })
-
-
- await db.put({id: 1, time: new Date().getTime(), body: 'log AAAA' })
-
-
- await db.delete(1)
- })()

查询
查询有很多种情况,常见的 ORM 里提供范围查询和索引查询两种方法,范围查询中还可以分页查询。在 IndexedDB 中我们简化为 getByIndex。
查询需要使用到 IDBCursor 游标和 IDBIndex 索引。
- class Database {
- // ... 省略打开数据库的过程
- // constructor(options = {}) {}
- // createDB() {}
- // createStore() {}
-
- // 查询第一个 value 相匹对的值
- get (value, indexName) {
- return new Promise((resolve, reject) => {
- const db = this.db;
- const transaction = db.transaction('log', 'readwrite')
- const store = transaction.objectStore('log')
- let request
- // 有索引则打开索引来查找,无索引则当作主键查找
- if (indexName) {
- let index = store.index(indexName);
- request = index.get(value)
- } else {
- request = store.get(value)
- }
-
-
- request.onsuccess = evt => evt.target.result ?
- resolve(evt.target.result) : resolve(null)
- request.onerror = evt => reject(evt)
- });
- }
-
-
- /**
- * 条件查询,带分页
- *
- * @param {string} keyPath 索引名称
- * @param {string} keyRange 索引对象
- * @param {number} offset 分页偏移量
- * @param {number} limit 分页页码
- */
- getByIndex (keyPath, keyRange, offset = 0, limit = 100) {
- return new Promise((resolve, reject) => {
- const db = this.db;
- const transaction = db.transaction('log', 'readonly')
- const store = transaction.objectStore('log')
- const index = store.index(keyPath)
- let request = index.openCursor(keyRange)
- const result = []
- request.onsuccess = function (evt) {
- let cursor = evt.target.result
- // 偏移量大于 0,代表需要跳过一些记录
- if (offset > 0) {
- cursor.advance(offset);
- }
- if (cursor && limit > 0) {
- console.log(1)
- result.push(cursor.value)
- limit = limit - 1
- cursor.continue()
- } else {
- cursor = null
- resolve(result)
- }
- }
- request.onerror = function (evt) {
- console.err('getLogByIndex onerror', evt)
- reject(evt.target.error)
- }
-
-
- transaction.onerror = function(evt) {
- reject(evt.target.error)
- };
- })
- }
- }
-
-
- (async function() {
- const db = new Database({ name: 'db_test' })
- await db.createDB()
-
- await db.add({time: new Date().getTime(), body: 'log 1' })
- // {id: 1, time: new Date().getTime(), body: 'log 2' }
-
-
- await db.add({time: new Date().getTime(), body: 'log 2' })
-
-
- const time = new Date().getTime()
-
-
- await db.put({id: 1, time: time, body: 'log AAAA' })
-
-
- await db.add({time: new Date().getTime(), body: 'log 3' })
-
-
- // 查询最小是这个时间的的记录
- const test = await db.getByIndex('time', IDBKeyRange.lowerBound(time))
- // multi index query
- // await db.getByIndex('time, test_id', IDBKeyRange.bound([0, 99],[Date.now(), 2100]);)
-
-
- console.log(test)
- // 0: {id: 1, time: 1648453268858, body: 'log AAAA'}
- // 1: {time: 1648453268877, body: 'log 3', id: 3}
- })()

查询当然还有更多可能,比如查询一张表全部的数据,或者是 count 获取这张表的记录数量等,留待读者们自行扩展。
优化
我们需要将 Model 和 Database 拆开来,上文 createDB 的时候做一些改进,类似 ORM 一样提供映射,以及基础的增删改查方法。
- class Database {
- constructor(options = {}) {
- if (typeof indexedDB === 'undefined') {
- throw new Error('indexedDB is unsupported!')
- }
- this.name = options.name
- this.db = null
- this.version = options.version || 1
- // this.upgradeFunction = option.upgradeFunction || function () {}
- this.modelsOptions = options.modelsOptions
- this.models = {}
- }
-
-
- createDB () {
- return new Promise((resolve, reject) => {
- indexedDB.deleteDatabase(this.name);
- const request = indexedDB.open(this.name);
- // 当数据库升级时,触发 onupgradeneeded 事件。升级是指该数据库首次被创建,或调用 open() 方法时指定的数据库的版本号高于本地已有的版本。
- request.onupgradeneeded = () => {
- const db = request.result;
-
-
- console.log('db onupgradeneeded')
-
-
- Object.keys(this.modelsOptions).forEach(key => {
- this.models[key] = new Model(db, key, this.modelsOptions[key])
- })
- };
-
-
- // 打开成功
- request.onsuccess = () => {
- console.log('db open onsuccess')
- console.log('addLog, deleteLog, clearLog, putLog, getAllLog, getLog')
- resolve(request.result)
- this.db = request.result
- };
- // 打开失败
- request.onerror = function(event) {
- console.log('db open onerror', event);
- reject(event)
- }
- })
- }
- }
-
-
- class Model {
- constructor(database, tableName, options) {
- this.db = database
- this.tableName = tableName
-
-
- if (!this.db.objectStoreNames.contains(tableName)) {
- const objectStore = this.db.createObjectStore(tableName, {
- keyPath: options.keyPath,
- autoIncrement: options.autoIncrement || false
- });
- options.index && Object.keys(options.index).forEach(key => {
- objectStore.createIndex(key, options.index[key]);
- })
- }
- }
-
-
- add(data) {
- // ... 省略上文的 add 函数
- }
- delete(id) {
- // ... 省略
- }
- put(data) {
- // ... 省略
- }
- getByIndex(keyPath, keyRange) {
- // ... 省略
- }
- get(indexName, value) {
- // ... 省略
- }
- }
调用如下:
- (async function() {
- const db = new Database({
- name: 'db_test',
- modelsOptions: {
- log: {
- keyPath: 'id',
- autoIncrement: true,
- rows: {
- id: 'number',
- time: 'number',
- body: 'string',
- },
- index: {
- time: 'time'
- }
- }
- }
- })
- await db.createDB()
-
-
- await db.models.log.add({time: new Date().getTime(), body: 'log 1' })
-
- await db.models.log.add({time: new Date().getTime(), body: 'log 2' })
-
- await db.models.log.get(null, 1)
-
-
- const time = new Date().getTime()
-
-
- await db.models.log.put({id: 1, time: time, body: 'log AAAA' })
-
-
- await db.models.log.getByIndex('time', IDBKeyRange.only(time))
- })()
当然这只是一个很简陋的模型,它还有一些不足。比如查询时,开发者调用时不需要接触 IDBKeyRange,类似是 sequelize 风格的,映射为 time: { $gt: new Date().getTime() },用 $gt 来替代 IDBKeyRange.lowerbound。
批量操作
值得一提的,IndexedDB 的操作性能和提交给它的事务多少有着紧密的关系,推荐尽可能使用批量插入。
批量操作,可以采取事件委托来避免产生许多的 request 的 onsuccess、onerror 事件。
- class Model {
- // ... 省略 construct
-
-
- bulkPut(datas) {
- if (!(datas && datas.length > 0)) {
- return Promise.reject(new Error('no data'))
- }
- return new Promise((resolve, reject) => {
- const db = this.db;
- const transaction = db.transaction('log', 'readwrite')
- const store = transaction.objectStore('log')
-
-
- datas.forEach(data => store.put(data))
-
-
- // Event delegation
- // IndexedDB events bubble: request → transaction → database.
- transaction.oncomplete = function() {
- console.log('add transaction complete');
- resolve()
- };
- transaction.onabort = function (evt) {
- console.error('add transaction onabort', evt);
- reject(evt.target.error)
- }
- })
- }
- }
性能探索
IndexedDB 的 插入耗时 与提交给它的 事务数量 有显著的关联。我们设置一组对照实验:
测试代码如下:
- const promises = []
- for (let index = 0; index < 1000; index++) {
- promises.push(db.models.log.add({time: new Date().getTime(), body: `log ${index}` }))
- }
- console.time('promises')
- Promise.all(promises).then(() => {
- console.timeEnd('promises')
- })
- // promises: 20837.403076171875 ms
- const arr = []
- for (let index = 0; index < 1000; index++) {
- arr.push({time: new Date().getTime(), body: `log ${index}` })
- }
- console.time('promises')
- await db.models.log.bulkPut(arr)
- console.timeEnd('promises')
- // promises: 250.491943359375 ms
减少事务提交非常重要,以至于需要有大量存入的操作时,都推荐日志在内存中尽可能合并下,再批量写入。
值得一提的是,body 在上面的对照实验中只写入了个位数的字符,假设每次写 5000 个字符,批量写入的时间也只是从 250ms 提升到 300ms,提升的并不明显。
让我们再来对比一组情况,我们会提交 1 个事务,插入 1000 条数据,在 0 到 500 万存量数据间进行测试,我们得到以下数据:
- for (let i = 0; i < 10000; i++) {
- let date = new Date()
- let datas = []
- for (let j = 0; j < 1000; j++) {
- datas.push({ time: new Date().getTime(), body: `log ${j}`})
- }
- await db.models.log.bulkPut(datas)
- datas = []
-
-
- if (i === 10 || i === 50
- || i === 100 || i === 500 || i === 1000 || i === 2000
- || i === 5000) {
- console.warn(`success for bulkPut ${i}: `, new Date() - date)
- } else {
- console.log(`success for bulkPut ${i}: `, new Date() - date)
- }
-
- }
-
-
- // success for bulkPut 10: 283
- // success for bulkPut 50: 310
- // success for bulkPut 100: 302
- // success for bulkPut 500: 296
- // success for bulkPut 1000: 290
- // success for bulkPut 2000: 150
- // success for bulkPut 5000: 201
上文数据表明波动并不大,给出结论在 500w 的数据范围内,插入耗时没有明显的提升。当然查询取决的因素更多,其耗时留待读者们自行验证。

对于 IndexedDB 来说,它只负责接收一个又一个的事务进行处理,而不管这些事务是从哪个 tab 页提交来的,就可能会产生多个 tab 页的 JS 程序往数据库里试图操作同一条数据的情况。
拿我们的 db 来举例,若我们修改创建 store 时的索引 time 为:
objectStore.createIndex('time', 'time', { unique: true });
同时打开 3 个 tab,每个 tab 都是每 20ms 往数据库里写入一份数据,大概率会出现 error,解决这个问题的理想方法是 SharedWorker API, SharedWorker 类似于 WebWorker,不同点在于 SharedWorker 可以在多个上下文之间共享。我们可以在 SharedWorker 中创建数据库,所有浏览器的 tab 都可以向 Worker 请求数据,而不是自己建立数据库连接。
遗憾的是 SharedWorker API 在 Safari 中无法支持,没有 polyfill。作为取代,我们可以使用 BroadcastChannel API,他可以在多 tab 间通信,选举出一个 leader,允许 leader 拥有写入数据库的能力,而其他 tab 只能读不能写。
下面是一个 leader 选举过程的简单代码,参照自 broadcast-channel。
- class LeaderElection {
- constructor(name) {
- this.channel = new BroadcastChannel(name)
- // 是否已经存在 leader
- this.hasLeader = false
- // 是否自己作为 leader
- this.isLeader = false
-
-
- // token 数,用于无 leader 时同时有多个 apply 的情况,来比对 maxTokenNumber 确定最大的作为 leader
- this.tokenNumber = Math.random()
- // 最大的 token,用于无 leader 时同时有多个 apply 的情况,来选举一个最大的作为 leader
- this.maxTokenNumber = 0
- this.channel.onmessage = (evt) => {
- console.log('channel onmessage', evt.data)
- const action = evt.data.action
- switch (action) {
- // 收到申请拒绝,或者是其他人已成为 leader 的宣告,则标记 this.hasLeader = true
- case 'applyReject':
- this.hasLeader = true
- break;
- case 'leader':
- // todo, 可能会产生另一个 leader
- this.hasLeader = true
- break;
- // leader 已死亡,则需要重新推举
- case 'death':
- this.hasLeader = false
- this.maxTokenNumber = 0
- // this.awaitLeadership()
- break;
- // leader 已死亡,则需要重新推举
- case 'apply':
- if (this.isLeader) {
- this.postMessage('applyReject')
- } else if (this.hasLeader) {
- } else if (evt.data.tokenNumber > this.maxTokenNumber) {
- // 还没有 leader 时,若自己 tokenNumber 比较小,那么记录 maxTokenNumber,
- // 将在 applyOnce 的过程中,撤销成为 leader 的申请。
- this.maxTokenNumber = evt.data.tokenNumber
- }
- break;
- default:
- break;
- }
- }
- }
- awaitLeadership() {
- return new Promise((resolve) => {
- const intervalApply = () => {
- return this.sleep(4000)
- .then(() => {
- return this.applyOnce()
- })
- .then(() => resolve())
- .catch(() => intervalApply())
- }
- this.applyOnce()
- .then(() => resolve())
- .catch(err => intervalApply())
- })
- }
- applyOnce(timeout = 1000) {
-
-
- return this.postMessage('apply').then(() => this.sleep(timeout))
- .then(() => {
- if (this.isLeader) {
- return
- }
- if (this.hasLeader === true || this.maxTokenNumber > this.tokenNumber) {
- throw new Error()
- }
- return this.postMessage('apply').then(() => this.sleep(timeout))
- })
- .then(() => {
- if (this.isLeader) {
- return
- }
- if (this.hasLeader === true || this.maxTokenNumber > this.tokenNumber) {
- throw new Error()
- }
- // 两次尝试后无人阻止,晋升为 leader
- this.beLeader()
- })
-
- }
- beLeader () {
- this.postMessage('leader')
- this.isLeader = true
- this.hasLeader = true
- clearInterval(this.timeout)
- window.addEventListener('beforeunload', () => this.die());
- window.addEventListener('unload', () => this.die());
- }
- die () {
- this.isLeader = false
- this.hasLeader = false
- this.postMessage('death')
- }
- postMessage(action) {
- return new Promise((resolve) => {
- this.channel.postMessage({
- action,
- tokenNumber: this.tokenNumber
- })
- resolve()
- })
- }
- sleep(time) {
- if (!time) time = 0;
- return new Promise(res => setTimeout(res, time));
- }
- }
调用代码如下:
- const elector = new LeaderElection('test_channel')
- window.elector = elector
- elector.awaitLeadership().then(() => {
- document.title = 'leader!'
- })

在浏览器中离线存放大量数据,我们目前只能使用 IndexedDB,使用 IndexedDB 会碰到几个问题: