小编给大家分享一下Rxjava2_Flowable_Sqlite_Android数据库访问的示例分析,相信大部分人都还不怎么了解,因此分享这篇文章给大家参考一下,希望大家阅读完这篇文章后大有收获,下面让我们一起去了解一下吧!
一、使用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数据库访问的示例分析”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注亿速云行业资讯频道!
亿速云「云数据库 MySQL」免部署即开即用,比自行安装部署数据库高出1倍以上的性能,双节点冗余防止单节点故障,数据自动定期备份随时恢复。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。