Rxjava - 我可以从不同的线程调用“Subject.onNext”方法吗？答案

【问题标题】：Rxjava - Can I invoke `Subject.onNext` method from different threads?Rxjava - 我可以从不同的线程调用“Subject.onNext”方法吗？
【发布时间】：2017-01-30 20:17:37
【问题描述】：

我有一个 Spring Web 服务器，我想为每 N 个（例如 10 个）请求我的控制器的客户端创建一个聊天室。

对服务器的每个请求都有自己的线程，如何收集每个 N 个请求并为其创建一个房间？我认为 Rxjava 对此有解决方案，那么我该如何实现它，如果我不能这样做，最好的解决方案是什么？

更新 1：

在@pavan-kumar 回答的帮助下，我创建了这个：

@RestController
public class GameController {

    private final PublishSubject<Integer> subject;
    private AtomicInteger counter = new AtomicInteger(0);

    @Autowired
    public GameController(PublishSubject<Integer> subject) {
        this.subject = subject;
    }

    @PostConstruct
    public void init() {


        subject.buffer(10).subscribe(
                integers -> {
                    StringBuilder builder = new StringBuilder("[ ");
                    for (Integer integer : integers) {
                        builder = builder.append(integer).append(", ");
                    }
                    String s = builder.append("]").toString();
                    System.out.println(s);
                });
    }


    @RequestMapping(value = "/game", method = RequestMethod.GET)
    public void findNewGame() {
        int i = counter.addAndGet(1);
        subject.onNext(i);

    }


}

所以当前的问题是“我可以从不同的线程调用Subject.onNext 方法吗？”

【问题讨论】：

标签： java spring multithreading rx-java

【解决方案1】：

不直接。您必须以某种方式提供序列化或使用toSerialized() 方法并与返回的Subject<T, R> 实例进行通信。

PublishSubject<Integer> ps = PublishSubject.create();

Subject<Integer, Integer> subject = ps.toSerialized();

subject.subscribe(System.out::println);

subject.onNext(1);

【讨论】：

【解决方案2】：

虽然用例对我来说不是很清楚，但以下方法可能会有所帮助。

Every request to server has its own thread - 使用在所有线程之间共享的static Observable。也许你可以在每次新用户建立连接时调用 onNext。

一旦您拥有这样的Observable，您就可以通过buffer 订阅它，如下所示。

Observable.range(1, 50).buffer(10).subscribe(n -> System.out.println(n.get(0)));

Observable.range(1, 50) 类似于您的static Observable，它会在每次建立连接时发出事件。 buffer 负责将所有项目合并到项目的List 中，并在发出所述数量的项目（10）时发出该列表。您可以在此subscribe 并根据需要采取适当的措施。

【讨论】：