前言
在正式讲解DataStore之前，我们先回顾下大家熟知的SharedPreferences（以下简称SP），众所周知SP有一些缺点，如调用getXXX()获取数据，可能会阻塞主线程；无法保证类型安全；加载的数据会一直留在内存中，浪费内存；apply方法无法获取到操作成功或失败的结果。
SharedPreferences回顾
getXXX()可能会阻塞主线程
请看以下SP获取value值的使用代码：
val sp = context.getSharedPreferences("sp_file", Context.MODE_PRIVATE)
sp.getString("name", "sp")看下getSharedPreferences的实现：
@Override
public SharedPreferences getSharedPreferences(File file, int mode) {
SharedPreferencesImpl sp;
synchronized (ContextImpl.class) {
final ArrayMap<File, SharedPreferencesImpl> cache = getSharedPreferencesCacheLocked();
sp = cache.get(file);
if (sp == null) {
checkMode(mode);
if (getApplicationInfo().targetSdkVersion >= android.os.Build.VERSION_CODES.O) {
...
}
sp = new SharedPreferencesImpl(file, mode);
cache.put(file, sp);
return sp;
}
}
if ((mode & Context.MODE_MULTI_PROCESS) != 0 ||
getApplicationInfo().targetSdkVersion < android.os.Build.VERSION_CODES.HONEYCOMB) {
sp.startReloadIfChangedUnexpectedly();
}
return sp;
}
里面调用了getSharedPreferencesCacheLocked方法获取一个以File为key的ArrayMap：
private ArrayMap<File, SharedPreferencesImpl> getSharedPreferencesCacheLocked() {
if (sSharedPrefsCache == null) {
sSharedPrefsCache = new ArrayMap<>();
}
final String packageName = getPackageName();
ArrayMap<File, SharedPreferencesImpl> packagePrefs = sSharedPrefsCache.get(packageName);
if (packagePrefs == null) {
packagePrefs = new ArrayMap<>();
sSharedPrefsCache.put(packageName, packagePrefs);
}
return packagePrefs;
}很简单，就是根据应用包名获取一个ArrayMap，并缓存到一个static的sSharedPrefsCache对象中。接着根据file获取对应的SharedPreferencesImpl，如果为空，则创建一个新的对象并缓存到内存中。
SharedPreferencesImpl构造方法会调用startLoadFromDisk方法，看下改方法实现：
private void startLoadFromDisk() {
synchronized (mLock) {
mLoaded = false;
}
new Thread("SharedPreferencesImpl-load") {
public void run() {
loadFromDisk();
}
}.start();
}启动一个子线层，调用了loadFromDisk方法，该方法就是读取对应SP文件，把键值对缓存到内存中。
接着看下getString方法的实现：
public String getString(String key, @Nullable String defValue) {
synchronized (mLock) {
awaitLoadedLocked();
String v = (String)mMap.get(key);
return v != null ? v : defValue;
}
}
private void awaitLoadedLocked() {
if (!mLoaded) {
BlockGuard.getThreadPolicy().onReadFromDisk();
}
while (!mLoaded) {
try {
mLock.wait();
} catch (InterruptedException unused) {
}
}
if (mThrowable != null) {
throw new IllegalStateException(mThrowable);
}
}看到会在一个同步块代码中调用awaitLoadedLocked方法，该方法调用了wait方法。如果读取一个较大的文件，这时又调用了getString方法，如果此时文件还未读取完成，则会造成主线程阻塞。
SP不能保证类型安全
请看以下代码：
val key = "name"
val sp = context.getSharedPreferences("sp_file", Context.MODE_PRIVATE)
sp.edit()?.apply {
putInt(key, 10)
}?.apply()
sp.getString(key, "sp")把一个Int类型的数据存储到SP中，紧接着调用getString获取同一个key的value值，这时变会出现类型转换异常。
apply方法可能会ANR
为啥apply方法会发生ANR呢，我们看下它的实现：
public void apply() {
final MemoryCommitResult mcr = commitToMemory();
final Runnable awaitCommit = new Runnable() {
@Override
public void run() {
try {
mcr.writtenToDiskLatch.await();
} catch (InterruptedException ignored) {
}
}
};
QueuedWork.addFinisher(awaitCommit);
Runnable postWriteRunnable = new Runnable() {
@Override
public void run() {
awaitCommit.run();
QueuedWork.removeFinisher(awaitCommit);
}
};
SharedPreferencesImpl.this.enqueueDiskWrite(mcr, postWriteRunnable);
notifyListeners(mcr);commitToMemory方法是把当前要提交的键值对缓存到内存，awaitCommit内调用了锁等待。每调用一次apply方法，就会把它添加到一个队列里。postWriteRunnable执行了awaitCommit的代码，接着移除当前队列里的awaitCommit。最后调用了enqueueDiskWrite方法。看下enqueueDiskWrite方法都做了什么：
private void enqueueDiskWrite(final MemoryCommitResult mcr,
final Runnable postWriteRunnable) {
final boolean isFromSyncCommit = (postWriteRunnable == null);
final Runnable writeToDiskRunnable = new Runnable() {
@Override
public void run() {
synchronized (mWritingToDiskLock) {
writeToFile(mcr, isFromSyncCommit);
}
synchronized (mLock) {
mDiskWritesInFlight--;
}
if (postWriteRunnable != null) {
postWriteRunnable.run();
}
}
};
if (isFromSyncCommit) {
boolean wasEmpty = false;
synchronized (mLock) {
wasEmpty = mDiskWritesInFlight == 1;
}
if (wasEmpty) {
writeToDiskRunnable.run();
return;
}
}
QueuedWork.queue(writeToDiskRunnable, !isFromSyncCommit);
}这里又有一个writeToDiskRunnable，里面执行了SP的持久化写入，同时调用传进来的postWriteRunnable。最后调用了QueuedWork.queue方法：
public static void queue(Runnable work, boolean shouldDelay) {
Handler handler = getHandler();
synchronized (sLock) {
sWork.add(work);
if (shouldDelay && sCanDelay) {
handler.sendEmptyMessageDelayed(QueuedWorkHandler.MSG_RUN, DELAY);
} else {
handler.sendEmptyMessage(QueuedWorkHandler.MSG_RUN);
}
}
}
private static class QueuedWorkHandler extends Handler {
static final int MSG_RUN = 1;
QueuedWorkHandler(Looper looper) {
super(looper);
}
public void handleMessage(Message msg) {
if (msg.what == MSG_RUN) {
processPendingWork();
}
}
}
private static void processPendingWork() {
synchronized (sProcessingWork) {
LinkedList<Runnable> work;
synchronized (sLock) {
work = sWork;
sWork = new LinkedList<>();
// Remove all msg-s as all work will be processed now
getHandler().removeMessages(QueuedWorkHandler.MSG_RUN);
}
if (work.size() > 0) {
for (Runnable w : work) {
w.run();
}
}
}
}queue方法把上面的runnable添加到一个队列中，这个队列里的任务会通过内部的HandlerThread执行添加到队列里的任务。看到这里也没发现会产生ANR。发现QueueWork里有一个waitToFinish方法：
/**
* Trigger queued work to be processed immediately. The queued work is processed on a separate
* thread asynchronous. While doing that run and process all finishers on this thread. The
* finishers can be implemented in a way to check weather the queued work is finished.
*
* Is called from the Activity base class's onPause(), after BroadcastReceiver's onReceive,
* after Service command handling, etc. (so async work is never lost)
*/
public static void waitToFinish() {
long startTime = System.currentTimeMillis();
boolean hadMessages = false;
Handler handler = getHandler();
synchronized (sLock) {
if (handler.hasMessages(QueuedWorkHandler.MSG_RUN)) {
// Delayed work will be processed at processPendingWork() below
handler.removeMessages(QueuedWorkHandler.MSG_RUN);
}
// We should not delay any work as this might delay the finishers
sCanDelay = false;
}
StrictMode.ThreadPolicy oldPolicy = StrictMode.allowThreadDiskWrites();
try {
processPendingWork();
} finally {
StrictMode.setThreadPolicy(oldPolicy);
}
try {
while (true) {
Runnable finisher;
synchronized (sLock) {
finisher = sFinishers.poll();
}
if (finisher == null) {
break;
}
finisher.run();
}
} finally {
sCanDelay = true;
}
synchronized (sLock) {
long waitTime = System.currentTimeMillis() - startTime;
if (waitTime > 0 || hadMessages) {
mWaitTimes.add(Long.valueOf(waitTime).intValue());
mNumWaits++;
}
}
}通过该方法的注释可以看出来，它会在Activity的onPause，BroadcastReceive的onReceive，Service的command handing调用。实际上是在handleServiceArgs、handleStopService、handlePauseActivity、handleStopActivity方法调用的，其实是一个意思。该方法有同步语句块，它等待队列里的任务都执行完成，如果执行大量的文件写入操作，就会造成对应生命周期组件发生ANR。
DataStore解析
初始化
使用由preferencesDataStore创建的委托属性来创建Datastore<Preference>实例。
preferencesDataStore委托给PreferenceDataStoreSingletonDelegate来实现：
internal class PreferenceDataStoreSingletonDelegate internal constructor(
private val name: String,
private val corruptionHandler: ReplaceFileCorruptionHandler<Preferences>?,
private val produceMigrations: (Context) -> List<DataMigration<Preferences>>,
private val scope: CoroutineScope
) : ReadOnlyProperty<Context, DataStore<Preferences>> {
private val lock = Any()
@GuardedBy("lock")
@Volatile
private var INSTANCE: DataStore<Preferences>? = null
override fun getValue(thisRef: Context, property: KProperty<*>): DataStore<Preferences> {
return INSTANCE ?: synchronized(lock) {
if (INSTANCE == null) {
val applicationContext = thisRef.applicationContext
INSTANCE = PreferenceDataStoreFactory.create(
corruptionHandler = corruptionHandler,
migrations = produceMigrations(applicationContext),
scope = scope
) {
applicationContext.preferencesDataStoreFile(name)
}
}
INSTANCE!!
}
}
}使用单例创建了一个DataStore<Preferences>实例并返回。name为文件名，produceMigrations用来做sp到datastore的迁移，scope协程作用域，在IO线程下创建。其中preferencesDataStoreFile(name)以name作为文件名创建了一个FIle，后缀名为.preferences_pb。
接着调用PreferenceDataStoreFactory工厂方法：
@JvmOverloads
public fun create(
corruptionHandler: ReplaceFileCorruptionHandler<Preferences>? = null,
migrations: List<DataMigration<Preferences>> = listOf(),
scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()),
produceFile: () -> File
): DataStore<Preferences> {
val delegate = DataStoreFactory.create(
serializer = PreferencesSerializer,
corruptionHandler = corruptionHandler,
migrations = migrations,
scope = scope
) {
val file = produceFile()
file
}
return PreferenceDataStore(delegate)
}这里根据delegate对象最后创建了DataStore<Preferences>实例。
看下DataStoreFactory工厂方法：
@JvmOverloads // Generate constructors for default params for java users.
public fun <T> create(
serializer: Serializer<T>,
corruptionHandler: ReplaceFileCorruptionHandler<T>? = null,
migrations: List<DataMigration<T>> = listOf(),
scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()),
produceFile: () -> File
): DataStore<T> =
SingleProcessDataStore(
produceFile = produceFile,
serializer = serializer,
corruptionHandler = corruptionHandler ?: NoOpCorruptionHandler(),
initTasksList = listOf(DataMigrationInitializer.getInitializer(migrations)),
scope = scope
)没有什么特殊操作，调用了SingleProcessDataStore来创建DataStore。看下PreferenceDataStore(delegate)：
internal class PreferenceDataStore(private val delegate: DataStore<Preferences>) :
DataStore<Preferences> by delegate {
override suspend fun updateData(transform: suspend (t: Preferences) -> Preferences):
Preferences {
return delegate.updateData {
val transformed = transform(it)
(transformed as MutablePreferences).freeze()
transformed
}
}
}将PreferenceDataStore对象代理给了传进来的delegate，并调用了updateData方法。在方法里将transformed转换成了MutablePreferences并调用了freeze方法。
读数据
看下SingleProcessDataStore的代码：
override val data: Flow<T> = flow {
val currentDownStreamFlowState = downstreamFlow.value
if (currentDownStreamFlowState !is Data) {
// We need to send a read request because we don't have data yet.
actor.offer(Message.Read(currentDownStreamFlowState))
}
emitAll(
downstreamFlow.dropWhile {
if (currentDownStreamFlowState is Data<T> ||
currentDownStreamFlowState is Final<T>
) {
// We don't need to drop any Data or Final values.
false
} else {
// we need to drop the last seen state since it was either an exception or
// wasn't yet initialized. Since we sent a message to actor, we *will* see a
// new value.
it === currentDownStreamFlowState
}
}.map {
when (it) {
is ReadException<T> -> throw it.readException
is Final<T> -> throw it.finalException
is Data<T> -> it.value
is UnInitialized -> error()
}
}
)
}实现了DataStore接口，对data进行了赋值。当创建DataStore实例时，currentDownStreamFlowState是一个UnInitialized类型，会调用actor.offer方法发送一个Message.Read用于读取文件里面的数据。看下Message接受消息的地方：
private val actor = SimpleActor<SingleProcessDataStore.Message<T>>(
scope = scope,
onComplete = {
it?.let {
downstreamFlow.value = Final(it)
}
synchronized(SingleProcessDataStore.activeFilesLock) {
SingleProcessDataStore.activeFiles.remove(file.absolutePath)
}
},
onUndeliveredElement = { msg, ex ->
if (msg is SingleProcessDataStore.Message.Update) {
msg.ack.completeExceptionally()
}
}
) { msg ->
when (msg) {
is SingleProcessDataStore.Message.Read -> {
handleRead(msg)
}
is SingleProcessDataStore.Message.Update -> {
handleUpdate(msg)
}
}
}这时msg是Message.Read，调用了handleRead(msg)方法：
private suspend fun handleRead(read: SingleProcessDataStore.Message.Read<T>) {
when (val currentState = downstreamFlow.value) {
is Data -> {
// We already have data so just return...
}
is ReadException -> {
if (currentState === read.lastState) {
readAndInitOrPropagateFailure()
}
}
UnInitialized -> {
readAndInitOrPropagateFailure()
}
is Final -> error("Can't read in final state.") // won't happen
}
}downstreamFlow.value为UnInitialized，接着调用readAndInitOrPropagateFailure()方法，该方法里面调用了readAndInit()方法：
private suspend fun readAndInit() {
val updateLock = Mutex()
var initData = readDataOrHandleCorruption()
val api = object : InitializerApi<T> {
override suspend fun updateData(transform: suspend (t: T) -> T): T {
return updateLock.withLock() {
val newData = transform(initData)
if (newData != initData) {
writeData(newData)
initData = newData
}
initData
}
}
}
initTasks?.forEach { it(api) }
initTasks = null // Init tasks have run successfully, we don't need them anymore.
updateLock.withLock {
initializationComplete = true
}
downstreamFlow.value = Data(initData, initData.hashCode())
}这里readDataOrHandleCorruption()方法调用了readData()方法，该方法会读取文件输入流，并转换为Preferences：
private suspend fun readData(): T {
try {
FileInputStream(file).use { stream ->
return serializer.readFrom(stream)
}
} catch (ex: FileNotFoundException) {
if (file.exists()) {
throw ex
}
return serializer.defaultValue
}
}
override suspend fun readFrom(input: InputStream): Preferences {
val preferencesProto = PreferencesMapCompat.readFrom(input)
val mutablePreferences = mutablePreferencesOf()
preferencesProto.preferencesMap.forEach { (name, value) ->
PreferencesSerializer.addProtoEntryToPreferences(name, value, mutablePreferences)
}
return mutablePreferences.toPreferences()
}最后根据返回的initData数据创建了一个Data并赋值给了downstreamFlow.value，后续便可以在downstreamFlow读取到最新数据。
写数据
通过调用DataStore<Preferences>.edit方法将数据写入到文件里：
public suspend fun DataStore<Preferences>.edit(
transform: suspend (MutablePreferences) -> Unit
): Preferences {
return this.updateData {
// It's safe to return MutablePreferences since we freeze it in
// PreferencesDataStore.updateData()
it.toMutablePreferences().apply { transform(this) }
}
}调用了delegate.updateData方法：
override suspend fun updateData(transform: suspend (t: T) -> T): T {
val ack = CompletableDeferred<T>()
val currentDownStreamFlowState = downstreamFlow.value
val updateMsg =
SingleProcessDataStore.Message.Update(transform, ack, currentDownStreamFlowState, coroutineContext)
actor.offer(updateMsg)
return ack.await()
}通过actor.offer发送了一个Message.Update消息，收到消息后，调用了handleUpdate(msg)方法：
private suspend fun handleUpdate(update: SingleProcessDataStore.Message.Update<T>) {
update.ack.completeWith(
runCatching {
when (val currentState = downstreamFlow.value) {
is Data -> {
// We are already initialized, we just need to perform the update
transformAndWrite(update.transform, update.callerContext)
}
is ReadException, is UnInitialized -> {
if (currentState === update.lastState) {
// we need to try to read again
readAndInitOrPropagateAndThrowFailure()
// We've successfully read, now we need to perform the update
transformAndWrite(update.transform, update.callerContext)
} else {
throw (currentState as ReadException).readException
}
}
is Final -> throw currentState.finalException // won't happen
}
}
)
}currentState为Data，调用了transformAndWrite(update.transform, update.callerContext)方法：
private suspend fun transformAndWrite(
transform: suspend (t: T) -> T,
callerContext: CoroutineContext
): T {
val curDataAndHash = downstreamFlow.value as Data<T>
curDataAndHash.checkHashCode()
val curData = curDataAndHash.value
val newData = withContext(callerContext) { transform(curData) }
curDataAndHash.checkHashCode()
return if (curData == newData) {
curData
} else {
writeData(newData)
downstreamFlow.value = Data(newData, newData.hashCode())
newData
}
}如果当前Data和新的Data相等时，则直接返回，不等则调用writeData(newData)：
internal suspend fun writeData(newData: T) {
file.createParentDirectories()
val scratchFile = File(file.absolutePath + SCRATCH_SUFFIX)
try {
FileOutputStream(scratchFile).use { stream ->
serializer.writeTo(newData, SingleProcessDataStore.UncloseableOutputStream(stream))
stream.fd.sync().
}
if (!scratchFile.renameTo(file)) {
throw IOException()
}
} catch (ex: IOException) {
if (scratchFile.exists()) {
scratchFile.delete() // Swallow failure to delete
}
throw ex
}
}创建输出流，把新数据写入到文件中。
总结
本文分析了SharedPreferences的缺点，接着从源码角度分析了DataStore，进行了俩者的优缺点对比。这里放一张官方的对比图：
