system.reactive - 简单命令式任务的响应式(Reactive)方法-6ren

system.reactive - 简单命令式任务的响应式(Reactive)方法

转载作者：行者123 更新时间：2023-12-04 12:18:51

申请要求:

订阅两个事件流 A 和 B

对于每个 A 事件，一段时间后应该有相应的 B 事件

如果没有相应的 B 到达(及时)，应用程序会监视 A 事件并发出警报

B 事件可以以与 A 事件不同的顺序到达(但仅限于之后)、迟到或根本不到达

这在传统方法中很简单:

记录集合中的每个 A 事件

当对应的 B 事件到达时移除 A 事件

监视在超时内未收到 B 的 A 事件的集合，并生成警报

我想尝试一个响应式(Reactive)解决方案，但不知道哪个运营商最能表达这一点。我可视化事件流(Observables):

A流

B流

也许还有一个计时器滴答流(除非计时运算符满足此需求)

最终输出是一组警报:

someObservable.
    ...incantations, other observables...
.subscribe ((EventA a) -> raiseAlertForMissingB (a));

是否有一种优雅的响应式(Reactive)方法，或者这只是 不适合 rx ?

(对原始问题的后续补充)

ascii 艺术大理石图可能看起来像这样

A stream A1----A2----------A3------------------A4--------------A5---------
B stream ------------B1------------B3---------------B4------------B5-----
                                            (A2 TIMEOUT) 
merged   ------------A1B1----------A3B3------A2??---A4B4----------A5B5---

订阅者从合并的流中接收元组。如果元组是匹配的 A 和 B 事件，它会被记录，但如果元组是一个没有匹配 B 事件的 A 事件(在弹珠图中显示为 A2??)，订阅者会发出警报。中提琴!

但是如何为每个 A 事件触发匹配的 B 事件的定时等待？

(另一个补充)

为了说明第 4 个要求点部分“B 事件可以以与 A 事件不同的顺序到达(但仅在此之后)......”

A stream A6----A7----------A8----------------------
B stream ------------B7------------B6-------B8-----
merged   ------------A7B7----------A6B6-----A8B8---

2016 年 2 月 23 日更新

我正在测试移植到 Java 的建议解决方案。

解:谜

public class Test06Enigmativity {
    private static final long A_PERIOD = 400;
    private static final long B_PERIOD = 500;
    private static final int TIMEOUT = 4_000;
    private static final int[] bOrder = {
        0, 1, 2,
        4,
        3,  // out of order
        6,
        5,  // out of order
        7, 8,
        10, 11, 12, 13, 14,
        9,  // out of order
        15
    };

    private final long startTime = System.currentTimeMillis ();

    public static void main (final String[] args) {
        Test06Enigmativity app = new Test06Enigmativity ();
        app.runEnigmativity ();
    }

    private void runEnigmativity () {
        Observable<Long> aStream =
            Observable.interval (A_PERIOD, TimeUnit.MILLISECONDS)
                .doOnNext (seq -> {
                    output ("       A%s", seq);
                }).take (bOrder.length);

        Observable<Long> bStream =
            Observable.interval (B_PERIOD, TimeUnit.MILLISECONDS)
                .map (seq -> {
                    long aId = (long) bOrder[seq.intValue ()];
                    output ("           B%s", aId);
                    return aId;
                })
                .take (bOrder.length);

        monitorEnigmativity (aStream, bStream, TIMEOUT)
            .subscribe (this::output);

        try {
            Thread.sleep (60_000);
        } catch (InterruptedException e) {
        }
    }

    private Observable<String> monitorEnigmativity (Observable<Long> aStream, Observable<Long> bStream, int thresholdMsec) {
        return Observable.create (subscriber ->
            bStream.publish (pb ->
                Observable.merge (
                    aStream.map (ax ->
                        pb
                            .filter (pbx -> pbx.equals (ax))
                            .take (1)
                            .timeout (thresholdMsec, TimeUnit.MILLISECONDS, Observable.defer (
                                () -> {
                                    output ("           timeout on B%s", ax);
                                    return Observable.just (-1L);
                                }
                            )).map (pbx -> String.format ("%s,%s", ax, pbx))
                    )
                )
            ).subscribe (subscriber::onNext)
        );
    }

    private void output (String format, Object... args) {
        System.out.printf ("tid:%3d %4dms %s%n", Thread.currentThread ().getId (), System.currentTimeMillis () - startTime,
            String.format (format, args));
    }
}

这适用于:

private static final long A_PERIOD = 400;
private static final long B_PERIOD = 500;
private static final int TIMEOUT = 4_000;

tid: 12  511ms        A0
tid: 15  626ms            B0
tid: 15  626ms 0,0
tid: 12  907ms        A1
tid: 15 1126ms            B1
tid: 15 1126ms 1,1
tid: 12 1307ms        A2
tid: 15 1626ms            B2
tid: 15 1626ms 2,2
tid: 12 1707ms        A3
tid: 12 2107ms        A4
tid: 15 2126ms            B4
tid: 15 2126ms 4,4
tid: 12 2507ms        A5
tid: 15 2625ms            B3
tid: 15 2625ms 3,3
tid: 12 2907ms        A6
tid: 15 3126ms            B6
tid: 15 3126ms 6,6
tid: 12 3307ms        A7
tid: 15 3626ms            B5
tid: 15 3626ms 5,5
tid: 12 3707ms        A8
tid: 12 4107ms        A9
tid: 15 4126ms            B7
tid: 15 4126ms 7,7
tid: 12 4507ms        A10
tid: 15 4626ms            B8
tid: 15 4626ms 8,8
tid: 12 4908ms        A11
tid: 15 5127ms            B10
tid: 15 5128ms 10,10
tid: 12 5307ms        A12
tid: 15 5626ms            B11
tid: 15 5626ms 11,11
tid: 12 5707ms        A13
tid: 12 6107ms        A14
tid: 15 6126ms            B12
tid: 15 6126ms 12,12
tid: 12 6507ms        A15
tid: 15 6626ms            B13
tid: 15 6626ms 13,13
tid: 13 7109ms            timeout on B9
tid: 13 7114ms 9,-1
tid: 15 7126ms            B14
tid: 15 7126ms 14,14
tid: 15 7625ms            B9
tid: 15 8126ms            B15
tid: 15 8126ms 15,15

但是随着 B 的到达接近 A，事件可能会被错过:

private static final long A_PERIOD = 400;
private static final long B_PERIOD = 410;
private static final int TIMEOUT = 2_000;

tid: 12  509ms        A0
tid: 15  538ms            B0
tid: 12  905ms        A1
tid: 15  948ms            B1
tid: 15  948ms 1,1
tid: 12 1305ms        A2
tid: 15 1358ms            B2
tid: 15 1358ms 2,2
tid: 12 1706ms        A3
tid: 15 1768ms            B4
tid: 12 2105ms        A4
tid: 15 2178ms            B3
tid: 15 2178ms 3,3
tid: 12 2505ms        A5
tid: 16 2538ms            timeout on B0
tid: 16 2544ms 0,-1
tid: 15 2588ms            B6
tid: 12 2905ms        A6
tid: 15 2998ms            B5
tid: 15 2998ms 5,5
tid: 12 3305ms        A7
tid: 15 3408ms            B7
tid: 15 3408ms 7,7
tid: 12 3705ms        A8
tid: 15 3817ms            B8
tid: 15 3817ms 8,8
tid: 12 4105ms        A9
tid: 14 4106ms            timeout on B4
tid: 14 4106ms 4,-1
tid: 15 4228ms            B10
tid: 12 4505ms        A10
tid: 15 4638ms            B11
tid: 12 4905ms        A11
tid: 16 4906ms            timeout on B6
tid: 16 4906ms 6,-1
tid: 15 5048ms            B12
tid: 12 5305ms        A12
tid: 15 5457ms            B13
tid: 12 5705ms        A13
tid: 15 5868ms            B14
tid: 12 6106ms        A14
tid: 13 6107ms            timeout on B9
tid: 13 6107ms 9,-1
tid: 15 6279ms            B9
tid: 14 6510ms            timeout on B10
tid: 12 6510ms        A15
tid: 14 6510ms 10,-1
tid: 15 6688ms            B15
tid: 15 6688ms 15,15

B0 到达但产生超时而不是匹配。这是因为 B0 事件在 A-stream 的观察者改变订阅 B-stream 之前到达。我认为对于热门 B 流，嵌套订阅的流行方法是有缺陷的。需要的是某种有限的最近重播流 - 一个主题的应用程序？

解决方案:Supertopi

package test;

import rx.Observable;
import java.util.concurrent.TimeUnit;

public class Test06Supertopi {
    private static final long A_PERIOD = 400;
    private static final long B_PERIOD = 500;
    private static final int TIMEOUT = 3_000;
    private static final int[] bOrder = {
        0, 1, 2,
        4,
        3,  // out of order
        6,
        5,  // out of order
        7, 8,
        10, 11, 12, 13, 14,
        9,  // out of order
        15
    };

    private final long startTime = System.currentTimeMillis ();

    public static void main (final String[] args) {
        Test06Supertopi app = new Test06Supertopi ();
        app.runSupertopi ();
    }

    private void runSupertopi () {
        Observable<Long> aStream =
            Observable.interval (A_PERIOD, TimeUnit.MILLISECONDS)
                .doOnNext (seq -> {
                    output ("       A%s", seq);
                }).take (bOrder.length);

        Observable<Long> bStream =
            Observable.interval (B_PERIOD, TimeUnit.MILLISECONDS)
                .map (seq -> {
                    long aId = (long) bOrder[seq.intValue ()];
                    output ("           B%s", aId);
                    return aId;
                })
                .take (bOrder.length);

        monitorSupertopi (aStream, bStream, TIMEOUT)
            .subscribe (this::output);

        try {
            Thread.sleep (60_000);
        } catch (InterruptedException e) {
        }
    }

    private Observable<String> monitorSupertopi (Observable<Long> aStream, Observable<Long> bStream, int thresholdMsec) {
        return Observable.create (subscriber -> {
            Observable<Long> a = aStream.publish ().refCount ();
            Observable<Long> b = bStream.publish ().refCount ();

            a.subscribe ((Long aId) -> {
                Observable.merge (
                    Observable.timer (thresholdMsec, TimeUnit.MILLISECONDS)
                        .doOnNext (x -> {
                            output ("           timeout on B%s", aId);
                        })
                        .map (x -> String.format ("%s,%s", aId, -1L)),
                    b.filter ((Long j) -> j.equals (aId))
                        .map ((Long pbx) -> String.format ("%s,%s", aId, pbx))
                ).take (1)
                .subscribe (subscriber::onNext);
            });
        });
    }

    private void output (String format, Object... args) {
        System.out.printf ("tid:%3d %4dms %s%n", Thread.currentThread ().getId (), System.currentTimeMillis () - startTime,
            String.format (format, args));
    }
}

工作正常:

private static final long A_PERIOD = 400;
private static final long B_PERIOD = 500;
private static final int TIMEOUT = 4_000;

tid: 14  522ms        A0
tid: 14  922ms        A1
tid: 16 1054ms            B0
tid: 16 1055ms 0,0
tid: 14 1322ms        A2
tid: 16 1555ms            B1
tid: 16 1555ms 1,1
tid: 14 1721ms        A3
tid: 16 2055ms            B2
tid: 16 2055ms 2,2
tid: 14 2122ms        A4
tid: 14 2522ms        A5
tid: 16 2555ms            B4
tid: 16 2555ms 4,4
tid: 14 2922ms        A6
tid: 16 3055ms            B3
tid: 16 3055ms 3,3
tid: 14 3322ms        A7
tid: 16 3555ms            B6
tid: 16 3555ms 6,6
tid: 14 3722ms        A8
tid: 16 4055ms            B5
tid: 16 4055ms 5,5
tid: 14 4122ms        A9
tid: 14 4522ms        A10
tid: 16 4555ms            B7
tid: 16 4555ms 7,7
tid: 14 4922ms        A11
tid: 16 5055ms            B8
tid: 16 5055ms 8,8
tid: 14 5322ms        A12
tid: 16 5555ms            B10
tid: 16 5556ms 10,10
tid: 14 5723ms        A13
tid: 16 6056ms            B11
tid: 16 6057ms 11,11
tid: 14 6122ms        A14
tid: 14 6522ms        A15
tid: 16 6555ms            B12
tid: 16 6555ms 12,12
tid: 16 7055ms            B13
tid: 16 7055ms 13,13
tid: 13 7125ms            timeout on B9
tid: 13 7125ms 9,-1
tid: 16 7555ms            B14
tid: 16 7555ms 14,14
tid: 16 8055ms            B9
tid: 16 8555ms            B15
tid: 16 8555ms 15,15

但是随着时间的推移，最初的 B 事件会收到两次。我还在调查这个。

private static final long A_PERIOD = 400;
private static final long B_PERIOD = 410;
private static final int TIMEOUT = 2_000;

tid: 14  539ms        A0
tid: 14  939ms        A1
tid: 16  983ms            B0
tid: 16  983ms 0,0
tid: 14 1339ms        A2
tid: 16 1393ms            B1
tid: 16 1393ms 1,1
tid: 14 1739ms        A3
tid: 16 1803ms            B2
tid: 16 1803ms 2,2
tid: 14 2139ms        A4
tid: 16 2213ms            B4
tid: 16 2213ms 4,4
tid: 14 2539ms        A5
tid: 16 2623ms            B3
tid: 16 2623ms 3,3
tid: 14 2939ms        A6
tid: 16 3032ms            B6
tid: 16 3032ms 6,6
tid: 14 3339ms        A7
tid: 16 3443ms            B5
tid: 16 3443ms 5,5
tid: 14 3739ms        A8
tid: 16 3852ms            B7
tid: 16 3852ms 7,7
tid: 14 4139ms        A9
tid: 16 4263ms            B8
tid: 16 4263ms 8,8
tid: 14 4539ms        A10
tid: 16 4672ms            B10
tid: 16 4672ms 10,10
tid: 14 4939ms        A11
tid: 16 5083ms            B11
tid: 16 5083ms 11,11
tid: 14 5339ms        A12
tid: 16 5493ms            B12
tid: 16 5493ms 12,12
tid: 14 5739ms        A13
tid: 16 5903ms            B13
tid: 16 5903ms 13,13
tid: 14 6139ms        A14
tid: 13 6140ms            timeout on B9
tid: 13 6140ms 9,-1
tid: 16 6313ms            B14
tid: 16 6313ms 14,14
tid: 14 6539ms        A15
tid: 14 6950ms            B0
tid: 14 7360ms            B1
tid: 14 7770ms            B2
tid: 14 8180ms            B4
tid: 13 8540ms            timeout on B15
tid: 13 8540ms 15,-1

24/2/2016

解决方案:李·坎贝尔

package test;

import rx.Observable;
import rx.subjects.PublishSubject;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class Test06LeeCampbell {
    private static final int TIMEOUT = 500;

    class ScheduledEvent {
        final String type;
        final long aId;
        final long atMsec;
        volatile boolean expired;

        public ScheduledEvent (long atMsec, String type, long aId) {
            this.atMsec = atMsec;
            this.type = type;
            this.aId = aId;
        }
    }

    ScheduledEvent[] scheduledEvents = {
        new ScheduledEvent (10, "A", 0),
        new ScheduledEvent (90, "B", 0),
        new ScheduledEvent (110, "A", 1),
        new ScheduledEvent (140, "B", 1),
        new ScheduledEvent (190, "A", 2),
        new ScheduledEvent (270, "B", 2),
        new ScheduledEvent (310, "A", 3),
        new ScheduledEvent (410, "A", 4),
        new ScheduledEvent (440, "B", 4),
        new ScheduledEvent (480, "B", 3),
        new ScheduledEvent (510, "A", 5),
        new ScheduledEvent (610, "A", 6),
        //new ScheduledEvent (670, "B", 6),
        new ScheduledEvent (710, "A", 7),
        new ScheduledEvent (810, "A", 8),
        new ScheduledEvent (860, "B", 7),
        new ScheduledEvent (880, "B", 8),
        new ScheduledEvent (910, "A", 9),
        new ScheduledEvent (1100, "A", 10),
        new ScheduledEvent (1110, "A", 11),
        new ScheduledEvent (1120, "A", 12),
        new ScheduledEvent (1130, "A", 13),
        new ScheduledEvent (1140, "A", 14),
        //new ScheduledEvent (1200, "B", 10),
        //new ScheduledEvent (1210, "B", 11),
        //new ScheduledEvent (1220, "B", 12),
        new ScheduledEvent (1230, "B", 13),
        new ScheduledEvent (1240, "B", 14),
        new ScheduledEvent (1390, "B", 9),
        new ScheduledEvent (1450, "A", 15),
        new ScheduledEvent (3290, "B", 5),
        new ScheduledEvent (3350, "B", 15)
    };

    private final long startTime = System.currentTimeMillis ();

    public static void main (final String[] args) {
        Test06LeeCampbell app = new Test06LeeCampbell ();
        app.runLeeCampbell ();
    }

    private void runLeeCampbell () {
        Observable<Long> aStream =
            getCrudeSequencer ("A")
                .doOnNext (seq -> {
                    output ("       A%s", seq);
                });

        Observable<Long> bStreamCold =
            getCrudeSequencer ("B")
                .doOnNext (seq -> {
                    output ("           B%s", seq);
                });
        PublishSubject<Long> bStream = PublishSubject.create ();
        bStreamCold.subscribe (bStream);

        monitorLeeCampbell (aStream, bStream, TIMEOUT)
            .subscribe (this::output);

        pause (10_000);
    }

    private Observable<String> monitorLeeCampbell (Observable<Long> aStream, Observable<Long> bStream, int thresholdMsec) {
        return aStream.flatMap (a ->
            bStream.filter (b -> b.equals (a))
                .map (b -> String.format ("%s,%s", a, b))
                .take (1)
                .timeout (thresholdMsec, TimeUnit.MILLISECONDS)
                .onErrorResumeNext (
                    throwable -> {
                        output ("           timeout on B%s", a);
                        if (!(throwable instanceof TimeoutException)) {
                            throw new RuntimeException (throwable);
                        }
                        return Observable.just (String.format ("%s,%s", a, -1L));
                    }
                )
        );
    }

    private void output (String format, Object... args) {
        System.out.printf ("tid:%3d %4dms %s%n", Thread.currentThread ().getId (), getElapsedMsec (),
            String.format (format, args));
    }

    private long getElapsedMsec () {
        return System.currentTimeMillis () - startTime;
    }

    private Observable<Long> getCrudeSequencer (String name) {
        return Observable.create (subscriber ->
            new Thread (() -> {
                for (ScheduledEvent se : scheduledEvents) {
                    if (se.type.equals (name)) {
                        while (getElapsedMsec () < se.atMsec) {
                            pause (1);
                        }
                        subscriber.onNext (Long.valueOf (se.aId));
                        se.expired = true;
                    } else {
                        // Timing is not reliable for sequencing two threads
                        while (!se.expired) {
                            pause (1);
                        }
                    }
                }
                subscriber.onCompleted ();
            }).start ()
        );
    }

    private static void pause (final int millis) {
        try {
            Thread.sleep (millis);
        } catch (InterruptedException e) {
        }
    }
}

有效

tid: 12   90ms        A0
tid: 11  157ms            B0
tid: 11  157ms 0,0
tid: 12  159ms        A1
tid: 11  160ms            B1
tid: 11  160ms 1,1
tid: 12  190ms        A2
tid: 11  270ms            B2
tid: 11  270ms 2,2
tid: 12  310ms        A3
tid: 12  410ms        A4
tid: 11  440ms            B4
tid: 11  440ms 4,4
tid: 11  480ms            B3
tid: 11  480ms 3,3
tid: 12  510ms        A5
tid: 12  610ms        A6
tid: 12  710ms        A7
tid: 12  810ms        A8
tid: 11  860ms            B7
tid: 11  860ms 7,7
tid: 11  880ms            B8
tid: 11  880ms 8,8
tid: 12  910ms        A9
tid: 15 1011ms            timeout on B5
tid: 15 1012ms 5,-1
tid: 12 1100ms        A10
tid: 12 1110ms        A11
tid: 16 1111ms            timeout on B6
tid: 16 1111ms 6,-1
tid: 12 1120ms        A12
tid: 12 1130ms        A13
tid: 12 1140ms        A14
tid: 11 1230ms            B13
tid: 11 1230ms 13,13
tid: 11 1240ms            B14
tid: 11 1240ms 14,14
tid: 11 1390ms            B9
tid: 11 1390ms 9,9
tid: 12 1450ms        A15
tid: 14 1601ms            timeout on B10
tid: 14 1601ms 10,-1
tid: 15 1611ms            timeout on B11
tid: 15 1611ms 11,-1
tid: 16 1621ms            timeout on B12
tid: 16 1621ms 12,-1
tid: 19 1951ms            timeout on B15
tid: 19 1951ms 15,-1
tid: 11 3290ms            B5
tid: 11 3350ms            B15

最佳答案

假设:

B流很热，

您有一个可以表示匹配或不匹配的返回类型(而不是使用将终止订阅的 OnError/Timeout)

那么这样就好了。

AStream.SelectMany(a =>
    BStream.Where(b => b == a)
        .Select(b => new MatchMade(a, b))
        .Take(1)
        .Timeout(matchTimeout)
        .Catch<TimeoutException>(ex=>Observable.Return(new NoMatchMade(a)))
)

关于system.reactive - 简单命令式任务的响应式(Reactive)方法，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/35529284/

文章推荐： r - R中lm函数的详细信息

文章推荐： cmusphinx - 在Sphinx4中使用德语词典和语言模型

文章推荐： r - 更改 jupyter notebook 警告语言

src/app/reactive-form/reactive-form.component.html:47:48 - error TS2531: Object is possibly 'null'. src/app/reactive-form/reactive-form.component.ts(Src/app/reactive-form/reactive-form.component.html:47:48-错误TS2531：对象可能为‘Null’。Src/app/reactive-form/reactive-form.component.ts)
类型‘AbstractControl’上不存在属性‘Controls’。
system.reactive - System.Reactive.Unit 不叫 System.Reactive.Void 有什么原因吗？
主要是我很好奇。我们有一个名为 Unit 的对象在我们的代码库中 - 代表桥梁或道路的组件。在我们的例子中，看到带有 Unit 的 ReactiveUI 命令可能会模棱两可。作为声明中的泛型之一。
reactive-programming - 如何做响应式(Reactive)六边形建筑
我一直听说六边形架构必须与任何框架无关，并使用接口(interface) (SPI) 来委托(delegate)不属于业务层的每个代码部分。但是如何在不使用额外框架的情况下通过六边形架构创建一个响应
reactive-programming - The Reactive Manifesto 中的事件驱动架构和消息驱动架构到底有什么区别？
我读了 Reactive Manifesto . 但我无法理解 event driven architectures 之间的核心差异和 message driven architectures .结果
system.reactive - 简单命令式任务的响应式(Reactive)方法
申请要求: 订阅两个事件流 A 和 B 对于每个 A 事件，一段时间后应该有相应的 B 事件如果没有相应的 B 到达(及时)，应用程序会监视 A 事件并发出警报 B 事件可以以与 A 事件不同的顺序
system.reactive - 用响应式(Reactive)编程画线的地方
Closed. This question is opinion-based。它当前不接受答案。想改善这个问题吗？更新问题，以便editing this post用事实和引用来回答。 4年前关闭。
reactive-cocoa - 模拟一个 Reactive？
我有一个 ViewModel，它在其初始化程序中有一个输入 init(sliderEvents: Reactive) { 在测试中我想做类似的事情 slider.send(.touchDownInsi
system.reactive - Reactive Extensions Switch() 在搜索中
经典实时搜索示例: var searchResults = from input in textBoxChanged from results in GetDa
reactive-programming - 响应式(Reactive)管道中的后台任务(即发即忘)
我有一个响应式(Reactive)管道来处理传入的请求。对于每个请求，我需要调用一个与业务相关的函数 ( doSomeRelevantProcessing )。完成后，我需要将发生的事情通知一些外部
system.reactive - 在 Reactive Extensions 中实现自定义的基于硬件计时器的调度程序
是否可以为响应式扩展实现基于硬件计时器的自定义调度程序？我该如何开始，有什么好的例子吗？我有一个硬件可以每毫秒向我发送一个准确的中断。我想利用它来创建更精确的 RX 调度程序。更新感谢 Asti
system.reactive - Rx 框架真的是 "Reactive"吗？
我正在通过网络浏览 Rx 框架 Material ，我发现了很多。现在，每当我为此在 google 上搜索时，我还会在 wikipedia 链接中找到“响应式(Reactive)编程”。由于响应式
reactive-programming - ReactiveX 被认为是响应式(Reactive)编程吗？
关闭。这个问题是opinion-based .它目前不接受答案。想改进这个问题？更新问题，以便 editing this post 可以用事实和引用来回答它. 6年前关闭。 Improve this
system.reactive - SignalR 与响应式(Reactive)扩展
SignalR 与响应式扩展是同一回事吗？你能解释一下为什么或为什么不吗？最佳答案不，它们绝对不是同一件事。 Reactive Extensions 是一个用于创建和组合可观察的数据流或事件流的库
system.reactive - 响应式(Reactive)框架/DoubleClick
我知道有一种简单的方法可以做到这一点 - 但今晚它打败了我...... 我想知道两个事件是否在 300 毫秒内发生，就像双击一样。在 300 毫秒内单击两次左键鼠标 - 我知道这是构建响应式(Rea
system.reactive - 部署利用 .Net Reactive Extensions
我们的应用程序使用 Reactive Extensions (Rx)。这些通常通过 Microsoft 的可下载包安装。但是，当我们发布应用程序时，我们会提供 dll 的副本(即 System.Cor
java - Reactive 和 Reactive Streams 有什么区别？
我想了解 Reactive 和 ReactiveStreams 之间的区别，特别是在 RxJava 的上下文中？我能想到的最多的是 Reactive Streams 在规范中有一些背压的概念，但它已
java - Quarkus 响应式(Reactive)客户端真的是响应式(Reactive)的吗？
我想探索来自 Quarkus 的响应式 REST 客户端的慢速后端，并在他们建议的样本 (https://github.com/quarkusio/quarkus-quickstarts/tree/m
reactive-programming - 使用响应式(Reactive) Couchbase java 驱动程序进行批处理
假设我有一个存储桶，我需要从中获取日期早于现在的文档。该文档如下所示: { id: "1", date: "Some date", otherObjectKEY: "key1" } 对于每个文档，我
system.reactive - 如何将带有 InvokeOperation 回调的此函数转换为 Reactive Extensions？
我有一个 RIA 服务数据服务，它有几个函数调用，如下所示: public InvokeOperation SomeFunc( SomeData data, Action> callb
system.reactive - 响应式(Reactive)扩展是否适合公共(public)汽车？
我一直在使用 Rx 在单个应用程序中创建事件总线(想想 CQRS/ES)，它似乎工作得很好。然而，在调查了一堆不同的事件溯源框架之后，我还没有看到使用过一次 Rx。与基于反射/容器的调度程序相比，它似

行者123

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

system.reactive - 简单命令式任务的响应式(Reactive)方法