una - mensaje de confirmacion java

Evento enviado por el servidor con Jersey: EventOutput no se cierra después de que el cliente abandona

Question

Estoy usando jersey para implementar un escenario SSE.

El servidor mantiene las conexiones vivas. Y envía datos a los clientes periódicamente.

En mi caso, existe un límite de conexión, solo un cierto número de clientes pueden suscribirse al servidor al mismo tiempo.

Entonces, cuando un nuevo cliente intenta suscribirse, hago una comprobación (EventOutput.isClosed) para ver si las conexiones anteriores ya no están activas, por lo que pueden dejar espacio para nuevas conexiones.

Pero el resultado de EventOutput.isClosed es siempre falso, a menos que el cliente llame explícitamente al cierre de EventSource. Esto significa que si un cliente se cae accidentalmente (corte de energía o corte de internet), todavía está acaparando la conexión, y los nuevos clientes no pueden suscribirse.

¿Hay una solución para esto?

Answer 1

@CuiPengFei,

Así que en mis viajes tratando de encontrar una respuesta a esto, me encontré con un repositorio que explica cómo manejar con elegancia las conexiones de los clientes desconectados.

Encapsula toda la lógica SSE EventOutput en un Service / Manager. En esto, activan un hilo que comprueba si EventOutput ha sido cerrado por el cliente. Si es así, cierran formalmente la conexión (EventOutput # close ()). Si no, intentan escribir en la transmisión. Si arroja una excepción, el cliente se desconecta sin cerrar y se encarga de cerrarlo. Si la escritura es exitosa, se devuelve el EventOutput al grupo ya que todavía es una conexión activa.

El repositorio (y la clase real) están disponibles aquí . También he incluido la clase sin importar a continuación en caso de que el repositorio se elimine alguna vez.

Tenga en cuenta que lo vinculan a un Singleton. La tienda debe ser única en el mundo.

public class SseWriteManager {

private final ConcurrentHashMap<String, EventOutput> connectionMap = new ConcurrentHashMap<>();

private final ScheduledExecutorService messageExecutorService;

private final Logger logger = LoggerFactory.getLogger(SseWriteManager.class);

public SseWriteManager() {
    messageExecutorService = Executors.newScheduledThreadPool(1);
    messageExecutorService.scheduleWithFixedDelay(new messageProcessor(), 0, 5, TimeUnit.SECONDS);
}

public void addSseConnection(String id, EventOutput eventOutput) {
    logger.info("adding connection for id={}.", id);
    connectionMap.put(id, eventOutput);
}

private class messageProcessor implements Runnable {
    @Override
    public void run() {
        try {
            Iterator<Map.Entry<String, EventOutput>> iterator = connectionMap.entrySet().iterator();
            while (iterator.hasNext()) {
                boolean remove = false;
                Map.Entry<String, EventOutput> entry = iterator.next();
                EventOutput eventOutput = entry.getValue();
                if (eventOutput != null) {
                    if (eventOutput.isClosed()) {
                        remove = true;
                    } else {
                        try {
                            logger.info("writing to id={}.", entry.getKey());
                            eventOutput.write(new OutboundEvent.Builder().name("custom-message").data(String.class, "EOM").build());
                        } catch (Exception ex) {
                            logger.info(String.format("write failed to id=%s.", entry.getKey()), ex);
                            remove = true;
                        }
                    }
                }
                if (remove) {
                    // we are removing the eventOutput. close it is if it not already closed.
                    if (!eventOutput.isClosed()) {
                        try {
                            eventOutput.close();
                        } catch (Exception ex) {
                            // do nothing.
                        }
                    }
                    iterator.remove();
                }
            }
        } catch (Exception ex) {
            logger.error("messageProcessor.run threw exception.", ex);
        }
    }
}

public void shutdown() {
    if (messageExecutorService != null && !messageExecutorService.isShutdown()) {
        logger.info("SseWriteManager.shutdown: calling messageExecutorService.shutdown.");
        messageExecutorService.shutdown();
    } else {
        logger.info("SseWriteManager.shutdown: messageExecutorService == null || messageExecutorService.isShutdown().");
    }

}} 

Answer 2

Quería proporcionar una actualización sobre esto:

Lo que sucedía es que eventSource en el lado del cliente (js) nunca entró en readyState ''1'' a menos que hiciéramos una transmisión tan pronto como se agregara una nueva suscripción. Incluso en este estado, el cliente podría recibir datos enviados desde el servidor. Agregar una llamada para transmitir un simple mensaje "OK" ayudó a activar eventSource en readyState 1.

Al cerrar la conexión desde el lado del cliente; ser proactivo en la limpieza de recursos, simplemente cerrar el evento en el lado del cliente no ayuda. Debemos hacer otra llamada ajax al servidor para forzar al servidor a realizar una transmisión. Cuando la transmisión es forzada, jersey limpiará las conexiones que ya no están activas y lanzará recursos de lanzamiento (Conexiones en CLOSE_WAIT). Si no, la conexión permanecerá en CLOSE_WAIT hasta que suceda la siguiente transmisión.