java - Rxjava-我可以从不同的线程调用 `Subject.onNext`方法吗？

我有一个Spring Web服务器，我想为每一个请求我的 Controller 的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方法吗？”

最佳答案

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

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

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

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

subject.onNext(1);