O AlgaSensors é um sistema distribuído para monitorar temperatura de sensores em tempo real. São vários serviços Java: manager, processor, monitor, generator e um client em Angular que se comunicam por RabbitMQ. O objetivo: fazer tudo conversar por eventos, sem acoplamento síncrono.

O fluxo: o processor recebe leituras de temperatura via HTTP, publica num exchange do RabbitMQ, e o monitor consome esses eventos para guardar histórico e disparar alertas.

Publicar e esquecer

O processor valida a leitura, monta o evento e joga num fanout exchange. Não espera ninguém processar.

@PostMapping(value = "/", consumes = MediaType.TEXT_PLAIN_VALUE)
public ResponseEntity<?> save(@PathVariable TSID sensorId, @RequestBody String input) {
    double temperature;
    try {
        temperature = Double.parseDouble(input);
    } catch (NumberFormatException e) {
        return ResponseEntity.badRequest().build();
    }

    var tlo = new TemperatureLogData(
        IdUtils.generateTimeBasedUUID(), sensorId, OffsetDateTime.now(), temperature
    );

    rabbitTemplate.convertAndSend(RabbitMQConfig.FANOUT_EXCHANGE_NAME, "", tlo, messagePostProcessor);
    return ResponseEntity.ok().build();
}

Escolhi fanout porque o processor não precisa saber quem está ouvindo. Hoje é o monitor, amanhã pode ser um serviço de relatório, e o processor não muda. Um detalhe: o convertAndSend exige uma routing key, mesmo que o fanout ignore. Ficou um "" ali.

O sensorId usa TSID, IDs ordenáveis por tempo. Melhor para paginar e ordenar do que UUID aleatório, e mais curto.

Um evento, dois consumidores

No monitor, o mesmo evento alimenta duas coisas: gravar histórico e checar alerta. Duas filas ligadas ao mesmo exchange:

@Bean
public Binding bindProcessTemperature() {
    return BindingBuilder.bind(processTemperatureQueue()).to(exchange());
}

@Bean
public Binding bindAlert() {
    return BindingBuilder.bind(alertQueue()).to(exchange());
}

Os dois listeners consomem em paralelo, com concorrência de 2 a 3 consumidores por fila:

@Transactional
@RabbitListener(queues = RabbitMQConfig.PROCESS_TEMPERATURE_QUEUE, concurrency = "2-3")
public void handleProcessTemperature(@Payload TemperatureLogData data) {
    temperatureMonitorService.handleProcessTemperature(data);
}

Dead letter queue

Mensageria assíncrona exige pensar no que acontece quando o processamento falha. A fila de temperatura tem uma dead letter queue: se a mensagem estourar, vai parar numa fila separada para inspeção.

Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", "");
arguments.put("x-dead-letter-routing-key", PROCESS_TEMPERATURE_DEAD_LETTER_QUEUE);

return QueueBuilder.durable(PROCESS_TEMPERATURE_QUEUE).withArguments(arguments).build();

Para confirmar, deixei uma armadilha: qualquer leitura com valor 10 lança exceção. Mando um 10, vejo a mensagem cair na DLQ.

A regra de alerta

Cada sensor pode ter um limite mínimo e máximo configurado. Se a leitura ultrapassa, dispara:

if (data.value().compareTo(alert.getMaxTemperature()) >= 0) {
    log.info("Alert MAX: SensorId {} Temp {}", data.sensorId(), data.value());
} else if (data.value().compareTo(alert.getMinTemperature()) <= 0) {
    log.info("Alert MIN: SensorId {} Temp {}", data.sensorId(), data.value());
}

Sensor sem alerta configurado? A leitura é ignorada. ifPresentOrElse resolve sem if aninhado.

Juntando tudo

Tudo roda com Docker Compose orquestrando RabbitMQ e os serviços, com depends_on e healthcheck para garantir que ninguém suba antes do broker estar de pé. O client em Angular consome a API do monitor e mostra gráficos de histórico e alertas recentes. Cada serviço escala e falha sozinho, sem derrubar o resto.