Lo que tengo.
Una aplicación FastAPI que consta de dos aplicaciones secundarias y están montadas y rutas específicas. Uno de ellos es una aplicación websocket. Un clúster de Kafka de un solo nodo con dos temas. Uno en el que escribe la aplicación websocket y otro en el que lee la aplicación websocket. Una aplicación de reacción que usa el cliente socket.io para conectarse a la aplicación FastAPI. Específicamente la aplicación secundaria websocket.
Lo que estoy tratando de lograr
Puedo escribir en el tema de Kafka desde la subaplicación websocket, ya que es sencillo. Cada usuario que se conecta a la aplicación recibe un socket. Estoy asignando cada ID de socket al identificador de usuario y almacenándolo para su uso posterior. Después de que el productor produce el mensaje, otra aplicación realiza otro procesamiento en los datos y luego escribe en un tema del que tengo que leer los mensajes. Básicamente un consumidor. En el mensaje, tengo un identificador de usuario con el que puedo detectar la identificación del socket.
Ahora, no estoy seguro de cómo y cuándo iniciar mi consumidor con respecto a una aplicación FastAPI. ¿Tiene que ser un módulo independiente? En caso afirmativo, ¿cómo devuelvo la respuesta al cliente websocket original? [Tengo una identificación de socket almacenada.]
En primer lugar, este es un problema común, porque efectivamente tiene un consumidor en un ciclo de sondeo que luego se publica una vez que recibe un evento, definitivamente puede afectar el rendimiento del ciclo de eventos dependiendo de la carga. Sin embargo, es factible, suponiendo que use las partes y piezas correctas.
En este ejemplo , podría volver a configurar el consumidor AIOKafka para publicar en el socket requerido. Tendrá que mantener una asignación de ID de socket a ID de registro (use redis).
PERO esto sigue agregando muchas cosas a un solo trabajador de API rápido (suponiendo que gunicon) y necesitaría observar los grupos de consumidores y las particiones con mucho cuidado para asegurarse de que los mensajes realmente lleguen al usuario correcto.
Personalmente, desacoplaría esto con un sistema separado para manejar el envío en tiempo real al cliente, ya sea con un sistema pub-sub de websocket alojado o usando algo como faust-streaming
Descargo de responsabilidad: Escribí el blog vinculado, trabajando en Ably.