为什么在与 PriceResource Publisher 的多个连接中，只有一个获得流？答案

【问题标题】：Why in multiple connections to PricesResource Publisher, only one gets the stream?为什么在与 PriceResource Publisher 的多个连接中，只有一个获得流？
【发布时间】：2020-05-18 18:57:13
【问题描述】：

似乎只有一个http客户端获取数据流，而其他客户端没有。

Publisher是不是热点数据，应该广播给所有订阅者？

package org.acme.kafka;

import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import org.jboss.resteasy.annotations.SseElementType;
import org.reactivestreams.Publisher;

import io.smallrye.reactive.messaging.annotations.Channel;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static io.reactivex.Flowable.fromIterable;

/**
 * A simple resource retrieving the "in-memory" "my-data-stream" and sending the items to a server sent event.
 */
@Path("/migrations")
public class StreamingResource {
    private volatile Map<String, String> counterBySystemDate = new ConcurrentHashMap<>();

    @Inject
    @Channel("migrations")
    Flowable<String> counters;

    @GET
    @Path("/stream")
    @Produces(MediaType.SERVER_SENT_EVENTS) // denotes that server side events (SSE) will be produced
    @SseElementType("text/plain") // denotes that the contained data, within this SSE, is just regular text/plain data
    public Publisher<String> stream() {
        Flowable<String> mainStream = counters.doOnNext(dateSystemToCount -> {
            String key = dateSystemToCount.substring(0, dateSystemToCount.lastIndexOf("_"));
            counterBySystemDate.put(key, dateSystemToCount);
        });
        return fromIterable(counterBySystemDate.values().stream().sorted().collect(Collectors.toList()))
                .concatWith(mainStream)
                .onBackpressureLatest();
    }
}

【问题讨论】：

你能给我们更多关于你在做什么的信息吗？代码sn -p，扩展名列表，你application.properties的内容？

标签： jax-rs rx-java rx-java2 resteasy quarkus

【解决方案1】：

您可以使用Replay 运算符或ConnectableObservable

【讨论】：

【解决方案2】：

然后我所做的是将 Flowable 上的消息注入 PublishSubject 管道并应用背压策略，从而提供下游广播。

【讨论】：