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

Question

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

服务器的每个请求都有自己的线程，如何收集每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 方法吗？”

Answer 1

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

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

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

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

subject.onNext(1);