如何对Flowable执行操作
Flowable 是一款开源的工作流引擎，它提供了一套完整的工作流管理解决方案。对于对Flowable执行操作，可以通过以下几个步骤来实现：
- 配置Flowable引擎：首先，需要在项目中引入Flowable的依赖，并配置Flowable引擎的相关参数，例如数据库连接信息、流程定义文件路径等。
- 流程定义：在Flowable中，流程是以BPMN 2.0标准进行建模的，因此需要使用BPMN 2.0规范的流程设计器进行流程的设计。设计完成后，将流程定义文件部署到Flowable引擎中。
- 启动流程实例：通过Flowable引擎的API，可以根据流程定义的Key启动一个流程实例。流程实例是根据流程定义创建的一个具体执行过程，它包含了当前流程的执行状态、流程变量等信息。
- 执行操作：在流程实例运行过程中，可以通过Flowable提供的API执行各种操作，例如提交任务、撤回任务、查询任务列表等。具体的操作方式取决于业务需求和流程设计。
- 完成任务：当一个任务被执行完成后，可以使用Flowable的API将任务标记为完成状态。在标记任务完成后，Flowable会根据流程定义中的流程路由条件继续执行后续任务或结束流程实例。
总之，对Flowable执行操作的过程主要涉及配置引擎、流程定义、启动流程实例、执行操作和完成任务等步骤。Flowable提供了丰富的API和工具来支持工作流的各种操作。作为一个云计算领域的专家和开发工程师，你可以利用Flowable来构建和管理复杂的工作流系统，实现高效的业务流程管理。在腾讯云上，可以使用腾讯云的Serverless Workflow服务来实现类似的工作流管理功能。Serverless Workflow是一项托管式的工作流引擎服务，可帮助开发者轻松构建、运行和编排复杂的工作流。详情请参考腾讯云Serverless Workflow产品介绍：Serverless Workflow。
相关·内容
1回答
我如何中断一个长期的任务？
public Flowable<Integer> simple(int number) {
Flowable<Integer> flowable = Flowable.create(emitter -> {
for (int i = 1; i <= number; ++i) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
浏览 0提问于2017-02-04得票数 0
回答已采纳
我想创建一个Flowable来流式传输从一个aynchronous REST API端点接收到的结果，该端点发送每个块的结果(有必要发送另一个请求，以使用前一个响应中的头部获取剩余的结果列表)。它不想使用反压策略，而是只在必要时(在Flowable中被请求时)等待发送后续的HTTP请求。
函数Flowable.generate非常接近我想要的，但它似乎只适用于同步调用。有没有像这样的异步调用解决方案，或者你有什么建议这样做吗？
谢谢
浏览 0提问于2017-08-25得票数 0
2回答
我有一个Flowable，它不断地发出项目，并且从不调用onError或onComplete。现在，我有了一个要与此Flowable合并的Completable，这样当Completable完成时，Flowable将调用onComplete。我不能直接更改提供给我的Flowable对象。 我遇到的一个问题是，我会在Flowable上使用takeUntil，但是Flowable可能会在任何时候停止发出项，我仍然希望Completable能够调用onComplete。 更新:因为我们可以做Completable.toFlowable()，所以我们可以合并两个Flowables。问题是，当其中一个
浏览 19提问于2019-01-15得票数 4
2回答
我正在使用RxJava3，并且我有下面的代码设置，我想要在中间，第一个和第二个可流动对象之间发出一个项目。有办法做到这一点吗？ firstFlowable.firstElement()
//I want to emit an item here
.flatMap { secondFlowable.firstElement() } 我这样做的原因是因为在firstFlowable初始化之后，secondFlowable初始化之前有很长一段时间，我想用一条消息通知UI操作已经开始，我正在等待来自secondFlowable的数据计算。 我尝试使用startWithItem，但这在一开始就初始化了
浏览 19提问于2021-04-30得票数 0
回答已采纳
我是RxJava2的新手，不确定是否正确使用它。
我有几个InputStreams正在从TCP流中读取数据，并且正在将数据读取到byte[]中。然后，我获取byte[]并为每个InputStream创建一个Flowable，然后将各个Flowable合并为一个。
我希望能够关闭1个InputStream，但仍然可以让合并的Flowable继续从未关闭的InputStream读取。
目前，我让它正确地从2 InputStreams读取，但当我调用close时，它导致了一个异常，java.net.SocketException: Socket closed被抛出。
我的问题是，我如何正确地关闭In
浏览 4提问于2017-07-12得票数 1
回答已采纳
我需要在收到事件(表示任何数据项rxJava)后的X秒内执行一些操作。但如果在X秒结束之前获得另一个事件，则需要在X秒内再次执行计时器重置和操作。 换句话说，我们有用户可以与之交互的UI元素。在用户对此UI元素保持不活动X秒后，我们需要执行一些操作。简单的解决方案可能如下所示： Disposable disp;
void init() {
final Flowable<Boolean> flowable =
mapDragOrZoomSubject.delay(TIME_X, TimeUnit.SECONDS)
浏览 31提问于2020-04-17得票数 0
回答已采纳
1回答
我正在尝试使用RxJava来理解ReactiveX，但是我不能理解整个反应式的想法。我的案例如下：
我要上Task课。它有一个执行HTTP请求并通过executeRequest()方法获得响应的perform()方法。可以多次执行该请求(定义的重复次数)。我想获取executeRequest()的所有结果，并将它们组合到Flowable数据流中，这样我就可以在perform()方法中返回此Flowable。所以最后，我希望我的方法返回Task执行的所有请求的结果。
executeRequest()返回Single，因为它只执行一个请求，并且可能只提供一个响应，或者根本不提供响应(在超时的情况下
浏览 0提问于2018-06-08得票数 1
我有一个可流动的in列表，这些in表示组中的用户in。
private val memberIdsFlowable: Flowable<List<String>> = groupFlowable
.map { it.members }
我要打个电话
fun fetchUserFlowable(id: String): Flowable<User>
对于列表中的每个id，并将它们组合成一个用户列表Flowable<List<User>>
我尝试的是这样的：
private val membersFlowable = me
浏览 0提问于2020-03-14得票数 0
在中是这样规定的：
默认情况下，可观察对象的运算符的缓冲区大小为128个元素(请参阅Flowable.bufferSize()，可以通过系统参数rx2.buffer- 全局覆盖该大小)。
我的问题是，如何访问和设置rx2.buffer-size？如果我这样做了：
import io.reactivex.Completable;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.Single;
import io.rea
浏览 80提问于2018-10-24得票数 0
我的目标是解开in列表的单项列表，并获得它的所有元素，以便将其放在列表中。
List<Single<List<Int>>> listOfSinglesOfListofInts = getListSingleListType(); // random method that gives us that.
List<Int> intList = new ArrayList<>();
我的目标是将所有Int内容从listOfSinglesOfListOfInts移动到listInt。这是我尝试过的：
ListOfSinglesOfListO
浏览 0提问于2018-10-30得票数 3
1回答
我感兴趣的是flatMap如何控制它的“子”线程，例如，下面的代码工作得很好：
private Flowable<PlcDataPackage> createIntervalPlcFlowable() {
return Flowable.interval(1, TimeUnit.SECONDS, Schedulers.computation())
.onBackpressureLatest()
.parallel()
.runOn(Schedulers.computation())
浏览 0提问于2018-11-30得票数 3
回答已采纳
1回答
下面是测试代码
final Flowable<Integer> f1 = Flowable.fromPublisher(s -> {
s.onNext(Integer.valueOf(1));
s.onComplete();
});
final Flowable<Integer> f2 = Flowable.fromPublisher(s -> {
s.onNext(Integer.valueOf(2));
s.onComplete();
});
F
浏览 0提问于2016-12-10得票数 0
1回答
RxJava可流动缓存到单个死锁
、、、、
下面是我的代码片段。 我知道你不应该像这样阻止cachedFlowable，但这只是一个例子。 它在blockingGet生产线上卡住了。 如果我用singleElement替换singleOrError，代码仍然会卡住。如果我用firstElement替换singleOrError，代码就不会再卡住了。 有人能给我解释一下为什么会这样吗？ public static void main(String[] args) {
final Flowable<Integer> cachedFlowable = Flowable.just(1).cache();
浏览 28提问于2019-10-19得票数 1
回答已采纳
我遇到了一些有线问题。当我第一次订阅时，它会进行网络调用，并将数据保存到数据库中，但loadFromDb()从不执行，也不会抛出任何错误。 这一切为什么要发生？ Flowable<Resource<List<List<DataSource>>>> getBoundResource(List<String> parentId) {
return new RxNetworkBoundResource<List<List<DataSource>>,
ContainerR
浏览 21提问于2019-03-17得票数 1
回答已采纳
然而，尝试在Kotlin中将重试添加到RxJava2 retrofit2调用中，无法找出正确的语法。
我做了一些研究，发现我需要使用类似于模式的retryWhen (用于重试3次)：
val api: Single<Item> = ...;
api
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.retryWhen {errors ->
errors
.zipWith(Observable.range(1
浏览 1提问于2018-09-04得票数 0
4回答
我知道我不能在查询中使用DELETE (顺便说一句，这很遗憾)，我会得到以下错误：
<i>Error:error: Observable query return type (LiveData, Flowable etc) can only be used with SELECT queries that directly or indirectly (via @Relation, for example) access at least one table.</i>
但是我不能使用@Delete(WHERE... xxx)，那么如何通过参数删除特定的行呢？
浏览 0提问于2017-11-29得票数 101
回答已采纳
我正在用Room实现一个Android应用程序。目前，我已经将Guave和Jetpack libaries包含在我的其他部分的项目中了。但是，如果这有助于解决我的问题，我不介意包含另一个库(即JxJava)。然而，我没有找到任何权威的文件来回答我的问题。
我需要对数据进行异步的、可观察的查询，并且观察处理程序必须在后台线程上运行，因为我必须对结果进行一些代价高昂的后处理。
对于Java来说，有两个如何与空间异步交互的选项:使用Rxjava或Guava/LiveData (参见)
如果我想要一个异步的、一次性的(而不是可观察的)查询并接受Guava/LiveData，那么API将返回一个List
浏览 6提问于2021-04-17得票数 0
回答已采纳
1回答
如何使用Flowable指定进程持续时间？如果从startEvent流逝，例如30天，过程将不会完成，它应该被关闭。我将Spring Boot与flowable结合使用。应该在图中指定还是在代码中指定？
浏览 3提问于2019-03-15得票数 0
我得到了这个随机异常，我的应用程序崩溃，同时进行网络调用与改造，并正在寻找一些解决方案的指导。
Fatal Exception: java.io.InterruptedIOException: thread interrupted
at okio.Timeout.throwIfReached(Timeout.java:145)
at okio.Okio$1.write(Okio.java:76)
at okio.AsyncTimeout$1.write(AsyncTimeout.java:180)
at okio.RealBuffered
浏览 1提问于2018-02-14得票数 16
回答已采纳
我创建了邮件任务配置
<serviceTask id="mailtask_name" name="Name" flowable:delegateExpression="${STCustomMail}" flowable:type="mail" >
<extensionElements>
<flowable:field name="to">
<flowable:string><![CDATA[mail@mail.com]]></
浏览 13提问于2019-04-01得票数 0
我试图用Rx方式使用room从数据库中检索数据。我就是这么做的
override fun onStart() {
super.onStart()
disposable.add(presenter.getAllBooks()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe({
println(it.size())
})
浏览 3提问于2017-11-13得票数 0
回答已采纳
我试图让下面的代码正常工作，但有些地方出了问题，下面是一个片段：
private void myMethod() {
Flowable.fromIterable(cache)
.zipWith(this::doesExist, (record, exist) -> {
// do stuff
return true;
}).subscrib();
}
private Flowable<Boolean> doesExist(CacheRecord record) {
// Do something
return Flowable.just(tr
浏览 0提问于2018-11-29得票数 0
回答已采纳
在rxjava (io.reactivex.rxjava2 - v2.0.8)中可观察到的流动和可观察的问题上，我遇到了奇怪的问题。这里有如下所示的代码，其中没有调用/执行map(...).subscribe(...)函数。
flowable.flatMap(...return new flowable...).map(...).subscribe(...)
令人惊讶的是，如果我翻转我的代码来使用可观察的而不是可流动的，那么map(...).subscribe(...)将按预期的方式被调用/执行。我可能错过了一些简单的东西，让我知道它可能是什么？
谢谢
`
public class Appli
浏览 1提问于2017-04-27得票数 4
1回答
我的存储库中有一个getPlaces方法：
override fun getPlaces(filter: FilterRequest): Flowable<List<Place>> {
return from(placesApi.filter(filter))
.doOnSuccess {
placesDao.savePlaces(it)
}
.flatMapPublisher { it ->
placesDao.getP
浏览 1提问于2018-06-12得票数 5
回答已采纳
1回答
我想在流上执行一些操作，然后将流分成两个流，然后分别处理它们。
显示问题的示例：
Flowable<SuccessfulObject> stream = Flowable.fromArray(
new SuccessfulObject(true, 0),
new SuccessfulObject(false, 1),
new SuccessfulObject(true, 2));
stream = stream.doOnEach(System.out::println);
Flowable<SuccessfulObject&
浏览 5提问于2017-05-04得票数 1
回答已采纳
1回答
在我的应用程序中，我正在为数据库层使用Room，而在网络调用中，我正在使用RxJava2 (这是我使用rxjava的第一个项目，所以我在这方面仍然很新手)。为了注入数据库、api等，我正在使用Dagger 2。
我想要进行一个网络调用并添加从网络到数据库的响应。当不需要进行另一个网络呼叫时，我想从数据库中获取数据。在我的房间存储库中使用可能/可流动的东西时，我有问题。
这是道：
@Dao
public interface CoinDao {
@Query("SELECT * FROM coin")
Flowable<List<Coin>>
浏览 0提问于2018-04-03得票数 11
回答已采纳
1回答
我使用Flowable.merge运算符，对来自两个上游的每个缓冲区大小为1。
Flowable.merge(sources, 2, 1)
.observeOn(Schedulers.io(), false, 1)
当从下游调用request(1)时，是否可以优先使用来自流的请求？当两个流都具有队列中的项(满)而不是应该发出的项时，第一个源的项(请求调用(1))。
来源可以是两个主题。
浏览 0提问于2018-04-30得票数 0
我需要从本地存储请求，如果它是空的，我应该请求在DB中服务和存储，但是如果DB不是空的，我应该忽略第二部分。我使用的是Room persistence DB (可流动的)，我的意思是如果我存储在DB中，我可以听到更改。我用的是凹式的，但相邻的部件都在工作
val item1 = itemDao.loadItem(id)
val item2 = apiIcaSeResource
.fetchItem(offerId)
.toFlowable()
.doOnNext { item ->itemDao.saveItem(Item(...) }
Flowable
浏览 3提问于2017-10-06得票数 1
回答已采纳
1回答
假设我在一个线程(不是main)上有以下数据库事务
query //asynchronous
insertions //synchronous
insertions //synchronous
根据我搜索的结果，查询是异步的，如果与其他同步事务顺序执行，它们会导致问题，甚至会从不同的线程并行插入。我如何控制哪个过程首先执行？
浏览 2提问于2018-06-07得票数 1
假设我有这样的模型类。
open class Person(
@PrimaryKey var id: Long = 0,
var name: String = "",
var age: Int = 0,
var dog: Dog? = null,
var cats: RealmList<Cat> = RealmList(),
) : RealmObject() {
}
有很多人，每个人都有一只或多只猫。
我的问题是如何获得Flowable<List<Cat>>
浏览 3提问于2017-11-01得票数 0
我有一个像这样组织的Single流：
getSomething() // returns Single<>
.flatMap(something -> {
// various things
return Single.defer( () -> {
// various other things
return Single.<SomeType>create(emitter -> {
// some more stuff
someCallbackApi(result -> {
浏览 19提问于2022-01-11得票数 1
回答已采纳
2回答
我有一个网络客户端，它能够从中断中恢复，但需要最后一条消息，这样做时，有重试。
以Kotlin为例：
fun requestOrResume(last: Message? = null): Flowable<Message> =
Flowable.create({ emitter ->
val connection = if (last != null)
client.start()
else
浏览 1提问于2017-10-30得票数 0
回答已采纳
我正在使用带有DROP BackPressure策略的RxJava 2 Flowable。有没有办法收集有关由于BackPressure而实际丢弃了多少消息的信息/统计数据？
解决方案
Flowable#onBackpressureDrop(consumer -> {});
浏览 18提问于2017-01-15得票数 2
回答已采纳
在RxJava2中，以下是我的代码：
public void myMethod() {
Flowable.create(e->{
// do sth.
}, BackpressureStrategy.BUFFER)
.subscribeOn(Schedulers.computation())
.observeOn(Schedulers.io())
.subscribe(new Subscriber<ContentsWithChannelVO>() {
@Override
public void onSubscribe(Subscription subscripti
浏览 0提问于2019-12-20得票数 2
我在科特林和RxJava一起工作。 我有一个函数，transform(f: Foo): Single<Bar> 如何获取Foo的List，并对每个next执行transform，以获得Flowable<Bar>的结果，其中Flowable中的每个next都是调用transform的结果 基本上，我需要一个函数 fun getFlowable(foos: List<Foo>): Flowable<Bar> {
// Runs transform() on each element of foos
// Concatenates t
浏览 8提问于2020-01-18得票数 0
回答已采纳
我已经退出Java游戏大约8年了，从那以后发生了很多变化。对我来说最大的挑战是RxJava / reactive。我正在寻找关于如何以一种完全反应的方式做下面的等价物的大致指导。 下面使用ThreadPoolExecutor实现的基本要求是通过调用远程web服务来处理大量Stuff，该服务的记录速率限制为每分钟100个请求。我的目标是在不丢失任何Stuff的情况下尽可能快地处理尽可能多的数据，但仍然遵守下行速率限制。此代码已简化，以避免错误、隔板、断路器、重试逻辑等。 这段代码目前工作得很好，但它的结果是，考虑到所有的非阻塞反应式选项，它会导致大量的线程浪费。甚至我用来调用我的服务的HTTP
浏览 24提问于2019-10-06得票数 1
回答已采纳
1回答
假设我在数据库里有一百万行记录。然后，我使用可流动(而不是可观察)读取记录，然后用缓慢的加密写入文件(写入要比读取慢得多)。
如果我在编写过程中不使用，那么是否有可能在编写过程尚在中途时，我的内存中填充了这100万条记录？换句话说，是否可能出现内存不足的异常？
如果是，是否使用“反应拉力”是防止这种异常的唯一方法，如果我也不想使用像节流这样的操作符，等等。
浏览 0提问于2017-09-05得票数 0
回答已采纳
我有一个像这样的函数：
public Flowable<Integer> max(int a, int b){
// *** Part 1 - start ***
int max = Math.max(a,b);
// *** Part 1 - end ***
return Flowable.defer(() -> {
// *** Part 2 start ***
return Flowable.just(max);
// *** Part 2 end ***
});
}
当我现在这
浏览 2提问于2018-02-06得票数 0
回答已采纳
我想分批处理一个Flowable，在每一批之后等待，直到一些异步工作在不阻塞线程的情况下完成，并以不同的方式处理最后一批。在onComplete()中，有比使用AtomicReference缓存前一批并处理最后一批更好的方法吗 AtomicReference<List<Integer>> batchRef = new AtomicReference<>();
Flowable.just(1, 2, 3, 4, 5, 6)
.buffer(2)
.concatMapComplet
浏览 35提问于2019-09-19得票数 1
假设我有一个Flowable列表，我想做一些事情直到所有的Flowable完成，我怎么做呢？
样本码
List<Flowable<Model>> flowableList = getFlowableList();
List<List<Model>> results = blockUntilAllFlowablesAreFinished(flowableList); // is there a similar method?
... // do something with results
浏览 1提问于2018-03-17得票数 1
回答已采纳
我正在尝试创建一个Flowable，它发布有关背压的事件，以避免内存问题，同时并行运行转换的每个阶段以提高效率。我创建了一个简单的测试程序来解释我的程序的不同步骤的行为以及事件何时发出和等待在不同的阶段。
我的计划如下：
public static void main(String[] args) throws ExecutionException, InterruptedException {
Stream<Integer> ints = IntStream.range(0, 1000).boxed().collect(Collectors.toList())
.
浏览 4提问于2017-07-18得票数 5
回答已采纳
1回答
将多个可观测数据连接到一个源
、、、、
我有很多不同类型的网络调用，比如字符串、int等等。
我在试着让它们平行。
在官方的rxjava文档中，我们可以读到：
Flowable.range(1, 10)
.parallel()
.runOn(Schedulers.computation())
.map(v -> v * v)
.sequential()
.blockingSubscribe(System.out::println);
这个例子很简单，因为我们的所有类型都是Int。但是，如果我们有不同的类型，例如字符串、布尔、Int，该如何做呢？
这个5的每个调用都是independentgroup，这5个调
浏览 3提问于2020-12-28得票数 0
2回答
使用RxJava2 RxKotlin和Room，我需要查询数据库以进行开放式搜索。这意味着我要搜索一个包含值为false的名为closed的属性的搜索。一旦找到猎物，它需要将查询切换到该特定猎物。
对于这些查询，我有两个方法：
getOpenHunt(teamId:String): Flowable<List<Hunt>>
getHunt(huntId:String): Flowable<List<Hunt>>
它们都返回一个List，否则当找不到猎物时，查询就会被卡住。
我的想法是这样的
fun queryHunt(teamId:String)
浏览 2提问于2017-10-02得票数 1
具有以下docker编写脚本 version: '3.1'
services:
flowable-ui:
image: flowable/flowable-ui
container_name: flowable-ui
depends_on:
- flowable-db
environment:
- SERVER_PORT=8888
- SPRING_DATASOURCE_DRIVER-CLASS_NAME=org.postgresql.Driver
- SPRING_DATASOURCE_U
浏览 38提问于2021-02-11得票数 0
我的视图模型调用存储库方法，从房间数据库和网络中获取一些数据。 class Repository @Inject constructor(
private val remoteDatasource: IRemoteSource,
private val localDatasource: ILocalSource,
private val subscriberScheduler: Scheduler,
private val observerScheduler: Scheduler
) : IRepository
浏览 11提问于2019-09-08得票数 0
回答已采纳
1回答
我正在尝试挂起流程实例，以便用户进行交互，以执行接受或拒绝流程流等操作。 public class UserWorkflowTaskListener implements ExecutionListener {
@Override
public void notify(DelegateExecution execution) {
runtimeService.suspendProcessInstanceById(execution.getProcessInstanceId());
}
} 但它给了我一个错误 2019-09-23 16:48:01.52
浏览 18提问于2019-09-23得票数 1
我刚刚在JElastic中安装了flowable/all- in -one docker镜像。当访问server.xxx/flowable-modeler时，会生成一个重定向http://localhost:8080/flowable-idm/#/login?redirectOnAuthSuccess=true&redirectUrl=http://server.xxx/flowable-modeler/ 该实例未在本地安装，因此浏览器可能找不到它。如何管理这些重定向？ 非常感谢
浏览 17提问于2019-09-27得票数 0
我的房间数据库中有3个方法
Flowable<List<Book>> getBooks() {....}
Flowable<List<Author>> getAuthors(int bookId) {....}
Flowable<List<Category>> getCategories(int bookId) {....}
Book对象
Book {int id; List<Author> authors; List<Category> categories}
我正在尝试做一些事情，比如
Flow
浏览 39提问于2018-08-09得票数 0
2回答
我希望将列表中的每个对象转换为另一个对象。但是在这样做的过程中，我的代码很难将它们转换回列表。
override fun myFunction(): LiveData<MutableList<MyModel>> {
return mySdk
.getAllElements() // Returns Flowable<List<CustomObject>>
.flatMap { Flowable.fromIterable(it) }
.map {MyModel(it.n
浏览 1提问于2018-11-15得票数 3
回答已采纳
我的数据库返回一个可流动的，如下所示：
Flowable<List<Users>> users = MyActivity.myDatabase.myDAO().getUsers();
我想快速、轻松地将它转换成一个列表，这样我就可以计算它，迭代它，或者将它作为一个ArrayList传递。使用Java做这件事的最好方法是什么？
浏览 2提问于2021-02-16得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
