RabbitMQ simplest Producer -> queue -> Consumer
Prerequisitos, tener corriendo el servidor rabbitMQ.referencias https://www.rabbitmq.com/getstarted.html
Crearemos este proyecto como maven, con dos modulos y un parent. un módulo sera el productor y el otro el consumidor. desde el proyecto parent maven podremos constuir ambos modulos (proyectos).
1.- Creamos el proyecto parent con nombre RabbitMQSimplest, el packaging se le indica como pom. el groupId, el artifactId, en este caso le indicaremos propiedades de que tipo de versión java usar.
<properties>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
y sus dependencias requeridas por rabbitMQ.
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.bext</groupId>
<artifactId>RabbitMQSimplest</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<artifactId>RabbitMQSimplestReceive</artifactId>
<properties>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
<dependencies>
<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.25</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.7.25</version>
</dependency>
</dependencies>
</project>
y así queda creado el proyecto parent.
2.- Creamos los dos subproyectos, como maven modules, sobre el parent proyect con new maven module, como cualquier otro proyecto.
De esta manera tenemos el proyecto del Productor y Consumidor agrupado. lo podemos correr desde el ide o desde linea de comandos.
bext@bext-VPCF13WFX:~/eclipse-workspace/RabbitMQSimplest$ mvn clean [INFO] Scanning for projects... [INFO] ------------------------------------------------------------------------ [INFO] Reactor Build Order: [INFO] [INFO] RabbitMQSimplest [pom] [INFO] RabbitMQSimplestSend [jar] [INFO] RabbitMQSimplestReceive [jar] [INFO] [INFO] ---------------------< com.bext:RabbitMQSimplest >---------------------- [INFO] Building RabbitMQSimplest 0.0.1-SNAPSHOT [1/3] [INFO] --------------------------------[ pom ]--------------------------------- [INFO] [INFO] --- maven-clean-plugin:2.5:clean (default-clean) @ RabbitMQSimplest --- [INFO] [INFO] -------------------< com.bext:RabbitMQSimplestSend >-------------------- [INFO] Building RabbitMQSimplestSend 0.0.1-SNAPSHOT [2/3] [INFO] --------------------------------[ jar ]--------------------------------- [INFO] [INFO] --- maven-clean-plugin:2.5:clean (default-clean) @ RabbitMQSimplestSend --- [INFO] Deleting /home/bext/eclipse-workspace/RabbitMQSimplest/RabbitMQSimplestSend/target [INFO] [INFO] ------------------< com.bext:RabbitMQSimplestReceive >------------------ [INFO] Building RabbitMQSimplestReceive 0.0.1-SNAPSHOT [3/3] [INFO] --------------------------------[ jar ]--------------------------------- [INFO] [INFO] --- maven-clean-plugin:2.5:clean (default-clean) @ RabbitMQSimplestReceive --- [INFO] Deleting /home/bext/eclipse-workspace/RabbitMQSimplest/RabbitMQSimplestReceive/target [INFO] ------------------------------------------------------------------------ [INFO] Reactor Summary for RabbitMQSimplest 0.0.1-SNAPSHOT: [INFO] [INFO] RabbitMQSimplest ................................... SUCCESS [ 0.293 s] [INFO] RabbitMQSimplestSend ............................... SUCCESS [ 0.006 s] [INFO] RabbitMQSimplestReceive ............................ SUCCESS [ 0.005 s] [INFO] ------------------------------------------------------------------------ [INFO] BUILD SUCCESS [INFO] ------------------------------------------------------------------------ [INFO] Total time: 0.535 s [INFO] Finished at: 2019-07-22T17:16:53-05:00 [INFO] ------------------------------------------------------------------------ bext@bext-VPCF13WFX:~/eclipse-workspace/RabbitMQSimplest$
bext@bext-VPCF13WFX:~/eclipse-workspace/RabbitMQSimplest$ mvn install [INFO] Scanning for projects... [INFO] ------------------------------------------------------------------------ [INFO] Reactor Build Order: [INFO] [INFO] RabbitMQSimplest [pom] [INFO] RabbitMQSimplestSend [jar] [INFO] RabbitMQSimplestReceive [jar] [INFO] [INFO] ---------------------< com.bext:RabbitMQSimplest >---------------------- [INFO] Building RabbitMQSimplest 0.0.1-SNAPSHOT [1/3] [INFO] --------------------------------[ pom ]--------------------------------- [INFO] [INFO] --- maven-install-plugin:2.4:install (default-install) @ RabbitMQSimplest --- [INFO] Installing /home/bext/eclipse-workspace/RabbitMQSimplest/pom.xml to /home/bext/.m2/repository/com/bext/RabbitMQSimplest/0.0.1-SNAPSHOT/RabbitMQSimplest-0.0.1-SNAPSHOT.pom [INFO] [INFO] -------------------< com.bext:RabbitMQSimplestSend >-------------------- [INFO] Building RabbitMQSimplestSend 0.0.1-SNAPSHOT [2/3] [INFO] --------------------------------[ jar ]--------------------------------- [INFO] [INFO] --- maven-resources-plugin:2.6:resources (default-resources) @ RabbitMQSimplestSend --- [WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent! [INFO] Copying 0 resource [INFO] [INFO] --- maven-compiler-plugin:3.1:compile (default-compile) @ RabbitMQSimplestSend --- [INFO] Nothing to compile - all classes are up to date [INFO] [INFO] --- maven-resources-plugin:2.6:testResources (default-testResources) @ RabbitMQSimplestSend --- [WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent! [INFO] Copying 0 resource [INFO] [INFO] --- maven-compiler-plugin:3.1:testCompile (default-testCompile) @ RabbitMQSimplestSend --- [INFO] Nothing to compile - all classes are up to date [INFO] [INFO] --- maven-surefire-plugin:2.12.4:test (default-test) @ RabbitMQSimplestSend --- [INFO] [INFO] --- maven-jar-plugin:2.4:jar (default-jar) @ RabbitMQSimplestSend --- [INFO] [INFO] --- maven-install-plugin:2.4:install (default-install) @ RabbitMQSimplestSend --- [INFO] Installing /home/bext/eclipse-workspace/RabbitMQSimplest/RabbitMQSimplestSend/target/RabbitMQSimplestSend-0.0.1-SNAPSHOT.jar to /home/bext/.m2/repository/com/bext/RabbitMQSimplestSend/0.0.1-SNAPSHOT/RabbitMQSimplestSend-0.0.1-SNAPSHOT.jar [INFO] Installing /home/bext/eclipse-workspace/RabbitMQSimplest/RabbitMQSimplestSend/pom.xml to /home/bext/.m2/repository/com/bext/RabbitMQSimplestSend/0.0.1-SNAPSHOT/RabbitMQSimplestSend-0.0.1-SNAPSHOT.pom [INFO] [INFO] ------------------< com.bext:RabbitMQSimplestReceive >------------------ [INFO] Building RabbitMQSimplestReceive 0.0.1-SNAPSHOT [3/3] [INFO] --------------------------------[ jar ]--------------------------------- [INFO] [INFO] --- maven-resources-plugin:2.6:resources (default-resources) @ RabbitMQSimplestReceive --- [WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent! [INFO] Copying 0 resource [INFO] [INFO] --- maven-compiler-plugin:3.1:compile (default-compile) @ RabbitMQSimplestReceive --- [INFO] Nothing to compile - all classes are up to date [INFO] [INFO] --- maven-resources-plugin:2.6:testResources (default-testResources) @ RabbitMQSimplestReceive --- [WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build is platform dependent! [INFO] Copying 0 resource [INFO] [INFO] --- maven-compiler-plugin:3.1:testCompile (default-testCompile) @ RabbitMQSimplestReceive --- [INFO] Nothing to compile - all classes are up to date [INFO] [INFO] --- maven-surefire-plugin:2.12.4:test (default-test) @ RabbitMQSimplestReceive --- [INFO] [INFO] --- maven-jar-plugin:2.4:jar (default-jar) @ RabbitMQSimplestReceive --- [INFO] [INFO] --- maven-install-plugin:2.4:install (default-install) @ RabbitMQSimplestReceive --- [INFO] Installing /home/bext/eclipse-workspace/RabbitMQSimplest/RabbitMQSimplestReceive/target/RabbitMQSimplestReceive-0.0.1-SNAPSHOT.jar to /home/bext/.m2/repository/com/bext/RabbitMQSimplestReceive/0.0.1-SNAPSHOT/RabbitMQSimplestReceive-0.0.1-SNAPSHOT.jar [INFO] Installing /home/bext/eclipse-workspace/RabbitMQSimplest/RabbitMQSimplestReceive/pom.xml to /home/bext/.m2/repository/com/bext/RabbitMQSimplestReceive/0.0.1-SNAPSHOT/RabbitMQSimplestReceive-0.0.1-SNAPSHOT.pom [INFO] ------------------------------------------------------------------------ [INFO] Reactor Summary for RabbitMQSimplest 0.0.1-SNAPSHOT: [INFO] [INFO] RabbitMQSimplest ................................... SUCCESS [ 0.502 s] [INFO] RabbitMQSimplestSend ............................... SUCCESS [ 1.235 s] [INFO] RabbitMQSimplestReceive ............................ SUCCESS [ 0.038 s] [INFO] ------------------------------------------------------------------------ [INFO] BUILD SUCCESS [INFO] ------------------------------------------------------------------------ [INFO] Total time: 1.981 s [INFO] Finished at: 2019-07-22T17:18:30-05:00 [INFO] ------------------------------------------------------------------------
El Productor
public class Send {
private final static String QUEUE_NAME = "simple-queue";
public static void main(String[] args) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try(Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
boolean durable = false;
boolean exclusive = false;
boolean autoDelete = false;
Map<String,Object> arguments = null;
channel.queueDeclare(QUEUE_NAME,durable, exclusive, autoDelete, arguments);
String message = "simple mensaje";
String exchange = "";
String routingKey = QUEUE_NAME;
AMQP.BasicProperties props = null;
channel.basicPublish(exchange, routingKey, props, message.getBytes());
System.out.println(" [x] enviado '" + message + "'");
}
}
}
En el código creamos una ConnectionFactory que implica la creación de sockets, negociación de protocolo según la versión, autenticación. Se conecta al host que tiene el broker rabbitmq, si esta en otra máquina se pondrá la respectiva IP.
Se crea un canal en un try-with-resources que cierran el canal implicitamente ya que la connexión y el canal implementan java.io.Closeable.
Declaramos un queue por el que publicaremos el mensaje. esto dentro del try.
la creación del queue es idempotente, así que se creara solo si no existe, el mensaje es un arreglo de bytes por lo que podemos meter lo que queramos.
El Consumidor
public class Receive {
public final static String QUEUE_NAME = "simple-queue";
public static void main(String[] args) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
boolean durable = false;
boolean exclusive = false;
boolean autoDelete = false;
Map<String,Object> arguments = null;
channel.queueDeclare(QUEUE_NAME, durable, exclusive, autoDelete, arguments);
System.out.println("Esperando por mensajes... CTRL+C para salir.");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("[x] recibido: '" + message +"'");
};
boolean autoAck = true;
channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag->{});
}
}
En el Consumidor hacemos lo mismo al principio que en el Productor, es decir creamos la conexión el canal, declaramos el mismo queue donde el productor deposita sus mensajes, y lo consumimos, indicandole una callback functión, que permite manejarlo asincronamente, cuyos parametros toman información del mensaje y con esta tomamos el cuerpo del mensaje y lo desplegamos.
La Prueba
Para hacer nuestra prueba, correremos tres consumidores, cada uno en una terminal, y despues en otra terminal enviaremos varios mensajes. Vemos que los consumidores van tomando alternadamente los mensajes.
En RabbitMQ Admin vemos que se crea el Queue, que hay tres canales y no un exchange.
Cuando terminamos El consumidor su conexión y canal se cierra.
RabbitMQSimplest$ sudo rabbitmqctl list_queues
Listing queues
simple-queue 0
eot
No hay comentarios:
Publicar un comentario