Multithreading – reactive extension onnext

Using the RX theme, is it thread safe to call onnext() from multiple threads?

Therefore, sequences can be generated from multiple sources

Do you do the same thing with merge?

Solution

RX contract requires notification to be continuous, which is a logical necessary condition for multiple operators That is, you can use the available synchronize methods to get this behavior

var subject = new Subject<int>();
        var syncedSubject = Subject.Synchronize(subject);

You can now make concurrent calls to syncedsubject For observers that must be synchronized, you can also use:

var observer = Observer.Create<Unit>(...);
        var syncedObserver = Observer.Synchronize(observer);

Test:

Func<int,Action> onNext = i => () => syncedSubject.OnNext(i);
        Parallel.Invoke
        (
            onNext(1),onNext(2),onNext(3),onNext(4)
        );