如何使用Rxjava执行房间交易

dependencies {
    implementation 'io.reactivex.rxjava3:rxjava:3.x.x'
}

Observable<Transaction> transactionStream = Observable.create(emitter -> {
    // 模拟房间交易事件的产生过程
    // 可以在这里调用相关的API获取房间交易数据
    // 将每个交易事件通过emitter发送出去
    emitter.onNext(new Transaction("房间A", "买入", 1000000));
    emitter.onNext(new Transaction("房间B", "卖出", 800000));
    emitter.onNext(new Transaction("房间C", "买入", 1200000));
    // ...
    emitter.onComplete();
});

transactionStream
    .filter(transaction -> transaction.getType().equals("买入"))
    .map(transaction -> {
        double amount = transaction.getAmount();
        double rmbAmount = amount * exchangeRate; // 假设有一个汇率exchangeRate
        return new Transaction(transaction.getRoom(), transaction.getType(), rmbAmount);
    })
    .subscribe(transaction -> {
        // 处理每个转换后的交易事件
        System.out.println("房间：" + transaction.getRoom());
        System.out.println("交易类型：" + transaction.getType());
        System.out.println("交易金额（人民币）：" + transaction.getAmount());
    });