无法从队列创建热流

Question

我有以下休息控制器，它接收请求，将它们转换为 JSON 字符串并将它们放入并发队列中。 我想从这个队列中创建一个 Flux 并订阅它。
不幸的是，它不起作用。
我在这里做错了什么？

@RestController
public class EventController {
    private final ObjectMapper mapper = new ObjectMapper();
    private final FirehosePutService firehosePutService;
    private ConcurrentLinkedQueue<String> events = new ConcurrentLinkedQueue<>();
    private int batchSize = 10;

    @Autowired
    public EventController(FirehosePutService firehosePutService) {
        this.firehosePutService = firehosePutService;

        Flux<String> eventFlux = Flux.create((FluxSink<String> sink) -> {
            String next;
            while (( next = events.poll()) != null) {
                sink.next(next);
            }
        });

        eventFlux.publish().autoConnect().subscribe(new BaseSubscriber<String>() {
            int consumed;
            List<String> batchOfEvents = new ArrayList<>(batchSize);

            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                request(batchSize);
            }

            @Override
            protected void hookOnNext(String value) {
                batchOfEvents.add(value);
                consumed++;

                if (consumed == batchSize) {
                    batchOfEvents.addAll(events);
                    log.info("Consume {} elements. Size of batchOfEvents={}", consumed, batchOfEvents.size());
                    firehosePutService.saveBulk(batchOfEvents);
                    consumed = 0;
                    batchOfEvents.clear();
                    events.clear();

                    request(batchSize);
                }
            }

        });
    }

    @GetMapping(value = "/saveMany", produces = "text/html")
    public ResponseEntity<Void> saveMany(@RequestParam MultiValueMap<String, String> allRequestParams) throws JsonProcessingException {
        Map<String, String> paramValues = allRequestParams.toSingleValueMap();
        String reignnEvent = mapper.writeValueAsString(paramValues);
        events.add(reignnEvent);

        return new ResponseEntity<>(HttpStatus.OK);
    }
}

Answer 1

首先，你使用poll 方法。如果队列为空，它不会阻塞并返回null。你循环收集直到第一个null（即while (next != null)，所以你的代码几乎立即退出循环，因为队列在开始时是空的。你必须用take替换poll，它是阻塞的，将等到元素可用。

其次，当事件从队列中移除时，hookOnNext 会被调用。但是，您正尝试使用batchOfEvents.addAll(events); 再次读取事件。此外，您还清除了所有未决事件events.clear();

我建议您从hookOnNext 方法中删除对events 集合的所有直接访问。

---

你为什么在这里使用 Flux？似乎过于复杂。你可以在这里使用普通线程

@Autowired
public EventController(FirehosePutService firehosePutService) {
    this.firehosePutService = firehosePutService;

    Thread persister = new Thread(() -> {
        List<String> batchOfEvents = new ArrayList<>(batchSize);

        String next;
        while (( next = events.take()) != null) {
            batchOfEvents.add(value);

            if (batchOfEvents.size() == batchSize) {
                log.info("Consume {} elements. Size of batchOfEvents={}", consumed, batchOfEvents.size());
                firehosePutService.saveBulk(batchOfEvents);
                batchOfEvents.clear();
            }
        }
    });
    persister.start();

}