一、使用Rxjava訪問數據庫的優點：
1.隨意的線程控制，數據庫操作在一個線程，返回數據處理在ui線程
2.隨時訂閱和取消訂閱，而不必再使用回調函數
3.對讀取的數據用rxjava進行過濾，流式處理
4.使用sqlbrite可以原生返回rxjava的格式，同時是響應式數據庫框架
（有數據添加和更新時自動調用之前訂閱了的讀取函數，達到有數據添加自動更新ui的效果，
同時這個特性沒有禁止的方法，只能通過取消訂閱停止這個功能，對于有的框架這反而是一種累贅）
二、接下來之關注實現過程：
本次實現用rxjava2的Flowable，有被壓支持（在不需要被壓支持的情況建議使用Observable）
實現一個穩健的的可靈活切換其他數據庫的結構，當然是先定義數據庫訪問接口。然后跟具不同的數據庫實現接口的方法
定義接口：（對于update，delete，insert，可以選擇void類型，來簡化調用代碼，但缺少了執行結果判斷）
public interface DbSource { //String sql = "insert into table_task (tid,startts) values(tid,startts)"; Flowable<Boolean> insertNewTask(int tid, int startts); //String sql = "select * from table_task"; Flowable<List<TaskItem>> getAllTask(); //String sql = "select * from table_task where endts = 0"; Flowable<Optional<TaskItem>> getRunningTask(); //String sql = "update table_task set isuploadend=isuploadend where tid=tid"; Flowable<Boolean> markUploadEnd(int tid, boolean isuploadend); //String sql = "delete from table_task where tid=tid and endts>0"; Flowable<Boolean> deleteTask(int tid); } 三、用Android原生的Sqlite實現數據庫操作
public class SimpleDb implements DbSource { private static SimpleDb sqlite; private SqliteHelper sqliteHelper; private SimpleDb(Context context) { this.sqliteHelper = new SqliteHelper(context); } public static synchronized SimpleDb getInstance(Context context) { if (sqlite == null ) sqlite = new SimpleDb(context); return sqlite; } Flowable<Boolean> insertNewTask(int tid, int startts) { return Flowable.create(new FlowableOnSubscribe<Boolean>() { @Override public void subscribe(FlowableEmitter<Boolean> e) throws Exception { //這里數據庫操作只做示例代碼，主要關注rxjava的Flowable使用方法 ContentValues values = new ContentValues(); values.put(“tid”， 1); values.put(“startts”，13233); if(sqliteHelper.getWriteableDatabase().insert(TABLE_NAME, null, values) != -1) e.onNext(true); else e.onNext(false); e.onComplete(); } }, BackpressureStrategy.BUFFER); } Flowable<List<TaskItem>> getAllTask() { return Flowable.create(new FlowableOnSubscribe<List<TaskItem>>() { @Override public void subscribe(FlowableEmitter<List<TaskItem>> e) throws Exception { List<TaskItem> taskList = new ArrayList<>(); StringBuilder sql = new StringBuilder(100); sql.append("select * from "); sql.append(SqliteHelper.TABLE_NAME_TASK); SQLiteDatabase sqLiteDatabase = sqliteHelper.getReadableDatabase(); Cursor cursor = sqLiteDatabase.rawQuery(sql.toString(), null); if (cursor.moveToFirst()) { int count = cursor.getCount(); for (int a = 0; a < count; a ++) { TaskItem item = new TaskItem(); item.setTid(cursor.getInt(1)); item.setStartts(cursor.getInt(2)); item.setEndts(cursor.getInt(3)); taskList.add(item); cursor.move(1); } } cursor.close(); sqLiteDatabase.close(); e.onNext(taskList); e.onComplete(); } }, BackpressureStrategy.BUFFER); } Flowable<Optional<TaskItem>> getRunningTask() { return Flowable.create(new FlowableOnSubscribe<Optional<TaskItem>>() { @Override public void subscribe(FlowableEmitter<Optional<TaskItem>> e) throws Exception { TaskItem item = null; StringBuilder sql = new StringBuilder(100); sql.append("select * from "); sql.append(SqliteHelper.TABLE_NAME_TASK); sql.append(" where endts=0 limit 1"); SQLiteDatabase sqLiteDatabase = sqliteHelper.getReadableDatabase(); Cursor cursor = sqLiteDatabase.rawQuery(sql.toString(), null); if (cursor.moveToFirst()) { int count = cursor.getCount(); if (count == 1) { item = new TaskItem(); item.setId(cursor.getInt(0)); item.setTid(cursor.getInt(1)); item.setStartts(cursor.getInt(2)); item.setEndts(cursor.getInt(3)); } } cursor.close(); sqLiteDatabase.close(); e.onNext(Optional.fromNullable(item)); //import com.google.common.base.Optional;//安全檢查，待會看調用的代碼，配合rxjava很好 e.onComplete(); } }, BackpressureStrategy.BUFFER); } Flowable<Boolean> markUploadEnd(int tid, boolean isuploadend) { return Flowable.create(new FlowableOnSubscribe<Boolean>() { @Override public void subscribe(FlowableEmitter<Boolean> e) throws Exception { //這里數據庫操作只做示例代碼，主要關注rxjava的Flowable使用方法 //數據庫操作代碼 e.onNext(false);//返回結果 e.onComplete();//返回結束 } }, BackpressureStrategy.BUFFER); } Flowable<Boolean> deleteTask(int tid) { return Flowable.create(new FlowableOnSubscribe<Boolean>() { @Override public void subscribe(FlowableEmitter<Boolean> e) throws Exception { //這里數據庫操作只做示例代碼，主要關注rxjava的Flowable使用方法 //數據庫操作代碼 e.onNext(false);//返回結果 e.onComplete();//返回結束 } }, BackpressureStrategy.BUFFER); } } 四、同一個接口使用sqlbrite的實現方式
public class BriteDb implements DbSource { @NonNull protected final BriteDatabase mDatabaseHelper; @NonNull private Function<Cursor, TaskItem> mTaskMapperFunction; @NonNull private Function<Cursor, PoiItem> mPoiMapperFunction; @NonNull private Function<Cursor, InterestPoiItem> mInterestPoiMapperFunction; // Prevent direct instantiation. private BriteDb(@NonNull Context context) { DbHelper dbHelper = new DbHelper(context); SqlBrite sqlBrite = new SqlBrite.Builder().build(); mDatabaseHelper = sqlBrite.wrapDatabaseHelper(dbHelper, Schedulers.io(); mTaskMapperFunction = this::getTask; mPoiMapperFunction = this::getPoi; mInterestPoiMapperFunction = this::getInterestPoi; } @Nullable private static BriteDb INSTANCE; public static BriteDb getInstance(@NonNull Context context) { if (INSTANCE == null) { INSTANCE = new BriteDb(context); } return INSTANCE; } @NonNull private TaskItem getTask(@NonNull Cursor c) { TaskItem item = new TaskItem(); item.setId(c.getInt(c.getColumnIndexOrThrow(PersistenceContract.TaskEntry.COLUMN_TASK_ID))); item.setTid(c.getInt(c.getColumnIndexOrThrow(PersistenceContract.TaskEntry.COLUMN_TASK_TID))); item.setStartts(c.getInt(c.getColumnIndexOrThrow(PersistenceContract.TaskEntry.COLUMN_TASK_STARTTS))); item.setEndts(c.getInt(c.getColumnIndexOrThrow(PersistenceContract.TaskEntry.COLUMN_TASK_ENDTS))); return item; } @Override public void insertNewTask(int tid, int startts) { ContentValues values = new ContentValues(); values.put(PersistenceContract.TaskEntry.COLUMN_TASK_TID, tid); values.put(PersistenceContract.TaskEntry.COLUMN_TASK_STARTTS, startts); mDatabaseHelper.insert(PersistenceContract.TaskEntry.TABLE_NAME_TASK, values, SQLiteDatabase.CONFLICT_REPLACE); } @Override public Flowable<List<TaskItem>> getAllTask() { String sql = String.format("SELECT * FROM %s", PersistenceContract.TaskEntry.TABLE_NAME_TASK);//TABLE_NAME_TASK表的名字字符串 return mDatabaseHelper.createQuery(PersistenceContract.TaskEntry.TABLE_NAME_TASK, sql) .mapToList(mTaskMapperFunction) .toFlowable(BackpressureStrategy.BUFFER); } @Override public Flowable<Optional<TaskItem>> getRunningTask() { String sql = String.format("SELECT * FROM %s WHERE %s = ? limit 1", PersistenceContract.TaskEntry.TABLE_NAME_TASK, PersistenceContract.TaskEntry.COLUMN_TASK_ENDTS); return mDatabaseHelper.createQuery(PersistenceContract.TaskEntry.TABLE_NAME_TASK, sql, "0") .mapToOne(cursor -> Optional.fromNullable(mTaskMapperFunction.apply(cursor))) .toFlowable(BackpressureStrategy.BUFFER); } @Override public Flowable<Boolean> markUploadEnd(int tid, boolean isuploadend) { return Flowable.create(new FlowableOnSubscribe<Boolean>() { @Override public void subscribe(FlowableEmitter<Boolean> e) throws Exception { ContentValues values = new ContentValues(); if(isuploadend) { values.put(PersistenceContract.TaskEntry.COLUMN_TASK_ISUPLOADEND, 1); } else { values.put(PersistenceContract.TaskEntry.COLUMN_TASK_ISUPLOADEND, 0); } String selection = PersistenceContract.TaskEntry.COLUMN_TASK_TID + " = ?"; //String[] selectionArgs = {String.valueOf(tid)}; String selectionArgs = String.valueOf(tid); int res = mDatabaseHelper.update(PersistenceContract.TaskEntry.TABLE_NAME_TASK, values, selection, selectionArgs); if (res > 0) { e.onNext(true);//返回結果 } else { e.onNext(false);//返回結果 } e.onComplete();//返回結束 } }, BackpressureStrategy.BUFFER); } @Override public Flowable<Boolean> deleteTask(int tid) { return Flowable.create(new FlowableOnSubscribe<Boolean>() { @Override public void subscribe(FlowableEmitter<Boolean> e) throws Exception { String selection = PersistenceContract.TaskEntry.COLUMN_TASK_TID + " = ? AND "+ PersistenceContract.TaskEntry.COLUMN_TASK_ENDTS + " > 0"; String[] selectionArgs = new String[1]; selectionArgs[0] = String.valueOf(tid); int res = mDatabaseHelper.delete(PersistenceContract.TaskEntry.TABLE_NAME_TASK, selection, selectionArgs); if (res > 0) { e.onNext(true);//返回結果 } else { e.onNext(false);//返回結果 } e.onComplete();//返回結束 } }, BackpressureStrategy.BUFFER); } } 五、數據庫調用使用方法
使用了lambda簡化了表達式進一步簡化代碼：
簡化方法：在/app/build.gradle里面加入如下內容：(defaultConfig的外面)
compileOptions { sourceCompatibility JavaVersion.VERSION_1_8 targetCompatibility JavaVersion.VERSION_1_8 } 接口調用（獲得數據庫實例）：
//全局定義的實例獲取類，以后想要換數據庫，只需在這個類里切換即可 public class Injection { public static DbSource getDbSource(Context context) { //choose one of them //return BriteDb.getInstance(context); return SimpleDb.getInstance(context); } } DbSource db = Injection.getInstance(mContext); disposable1 = db.getAllTask() .flatMap(Flowable::fromIterable) .filter(task -> { //自定義過濾 if (!task.getIsuploadend()) { return true; } else { return false; } }) .subscribe(taskItems -> //這里是使用了lambda簡化了表達式 doTaskProcess(taskItems) , throwable -> { throwable.printStackTrace(); },// onCompleted () -> { if (disposable1 != null && !disposable1.isDisposed()) { disposable1.dispose(); } }); disposable1 = db.getRunningTask() .filter(Optional::isPresent) //判斷是否為空，為空的就跳過 .map(Optional::get) //獲取到真的參數 .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(taskItem -> { //onNext() //has running task mTid = taskItem.getTid(); }, throwable -> throwable.printStackTrace() //onError() , () -> disposable1.dispose()); //onComplete() disposable1 = db.markUploadEnd(tid, isuploadend) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(status -> { //onNext() if (status) { //dosomething } }, throwable -> throwable.printStackTrace() //onError() , () -> disposable1.dispose()); //onComplete() disposable1 = db.deleteTask(tid) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(status -> { //onNext() if (status) { //dosomething } }, throwable -> throwable.printStackTrace() //onError() , () -> disposable1.dispose()); //onComplete() 以上這篇Rxjava2_Flowable_Sqlite_Android數據庫訪問實例就是小編分享給大家的全部內容了，希望能給大家一個參考，也希望大家多多支持武林網。
新聞熱點
疑難解答
