RabbitMQ Topics

RabbitMQ Topics

Referencia:https://www.rabbitmq.com/tutorials/tutorial-five-java.html
https://github.com/jalbertomr/RabbitMQTopics.git

  Declaramos el exchange como tipo topic, y practicamente nada más, la naturaleza de filtro del routingKey hace el trabajo por si solo.


public class EmitLogTopic {
    private static final String EXCHANGE_NAME = "topic_logs";
   
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
           
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
           
            String routingKey = getRoutingKey(args);
            String message = getMessage(args);
           
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
            System.out.println(" [x] Enviado '" + routingKey + "':'" + message + "'");
        };
    }

    private static String getMessage(String[] args) {
        return args[1];
    }

    private static String getRoutingKey(String[] args) {
        return args[0];
    }
}

..

public class ReceiveLogsTopic {
    private static final String EXCHANGE_NAME = "topic_logs";
   
    public static void main(String[] args) throws IOException, TimeoutException {
        if (args.length < 1) {
            System.err.println("Uso: ReceiveLogsTopic [binding key]...");
            System.exit(1);
        }
       
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
       
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();
       
        for (String bindingKey : args) {
            channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
        }
        System.out.println("[x] Esperando por mensajes. CTRL+C para salir.");
       
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("[x] recibido '" +
                    delivery.getEnvelope().getRoutingKey() +
                    "':'" + message + "'");
        };
        boolean autoack = true;
        channel.basicConsume(queueName, autoack, deliverCallback, consumerTag->{});
    }
}

Ejecución de ejemplo

Otro ejemplo con otra definicion de topicos

eot