使用 R2DBC 和 Reactor 从一个原始单声道生成许多其他单独的单声道

Question

我创建了一个增量为 5 的数据库序列，允许我在内存中保留一系列值，并将我对数据库的访问次数除以 5。（我的真实案例是 100000 个值）

我的 R2DBC 服务将调用 R2DBC API 并执行以单声道返回 long 的查询。

public class SequenceService {
   public Mono<Long> nextVal() {
      return r2dbc.query("select nextval('mysequence')");
   }
}

假设我们在测试中调用此服务

   sequenceService = new SequenceService( ... );

   assertTrue(sequenceService.nextVal().block() == 10L);
   assertTrue(sequenceService.nextVal().block() == 15L);
   assertTrue(sequenceService.nextVal().block() == 20L);

我想要做的是我的 nextVal 调用可以填充值之间，所以

   assertTrue(sequenceService().nextVal().block() == 10L);
   assertTrue(sequenceService().nextVal().block() == 11L);
   assertTrue(sequenceService().nextVal().block() == 12L);
   assertTrue(sequenceService().nextVal().block() == 13L);
   assertTrue(sequenceService().nextVal().block() == 14L);
   assertTrue(sequenceService().nextVal().block() == 15L);
   assertTrue(sequenceService().nextVal().block() == 16L);
   assertTrue(sequenceService().nextVal().block() == 17L);
   assertTrue(sequenceService().nextVal().block() == 18L);
   assertTrue(sequenceService().nextVal().block() == 19L);
   assertTrue(sequenceService().nextVal().block() == 20L);

像下面这样编码是行不通的，因为你总是会得到“n+0”，而永远不会得到“n+1”、“n+2”、“n+3”、“n+4”：

public class SequenceService {

    public Mono<Long> nextVal() {
        return r2dbc.query("select nextval('mysequence')")
                        .map(n -> Flux.range(0,5)).map(i -> n + i)).repeat().next();
        }
}

我不知道如何从一个原始单声道中生成许多其他单独的单声道。

我必须保留某种状态，即通量、当前序列值的状态，但不知道该怎么做。

Answer 1

一种可能性是创建一个将生成下一个值的 Flux，就像您所做的那样：

    r2dbc.query("select nextval('mysequence')")
        .flatMapMany(next -> Flux.range(0, 5).map(n -> next + n))
        .repeat()

然后立即订阅此通量，使用自定义订阅者，每次您需要下一个值时，该订阅者将使用 request(1) 方法按需请求下一个值。

但是，因为您的方法nextVal() 可能会被同时调用，所以您需要一种机制，例如等待列表。

这是一个完整的例子：

public class SequenceService {
        
    private Subscription subscription;
    private LinkedList<MonoSink<Long>> waitingList = new LinkedList<>();
    private Throwable error;
        
    public SequenceService() {
        r2dbc.query("select nextval('mysequence')")
            .flatMapMany(next -> Flux.range(0, 5).map(n -> next + n))
            .repeat()
            .subscribe(new Subscriber<Long>() {
                @Override
                public void onSubscribe(Subscription s) {
                    subscription = s;
                }
    
                @Override
                public void onNext(Long val) {
                    boolean needRequest;
                    MonoSink<Long> sink;
                    synchronized (waitingList) {
                        sink = waitingList.poll();
                        needRequest = !waitingList.isEmpty();
                    }
                    if (sink != null) {
                        sink.success(val);
                    }
                    if (needRequest)
                        subscription.request(1);
                }
    
                @Override
                public void onError(Throwable t) {
                    error = t;
                    synchronized (waitingList) {
                        MonoSink<Long> sink;
                        while ((sink = waitingList.poll()) != null)
                            sink.error(t);
                    }
                }
    
                @Override
                public void onComplete() {
                }
            });
    }
    
    public Mono<Long> nextVal() {
        return Mono.create(sink -> {
            boolean needRequest;
            synchronized (waitingList) {
                if (error != null) {
                    sink.error(error);
                    return;
                }
                waitingList.push(sink);
                needRequest = waitingList.size() == 1;
            }
            if (needRequest)
                subscription.request(1);
        });
    }
}