CloudAMQP with Java

We recommend the RabbitMQ Java client, maintained by the RabbitMQ team. It reconnects and recovers consumers automatically.

<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.37.0</version>
</dependency>

Publish and consume

Set CLOUDAMQP_URL to the AMQP URL from your instance’s details page in the CloudAMQP console. Without it, the examples connect to a local broker.

Save the connection helper as src/main/java/Amqp.java in your Maven project:

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Amqp {
  public static Connection connect() throws Exception {
    String uri = System.getenv().getOrDefault("CLOUDAMQP_URL", "amqp://guest:guest@localhost");
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri(uri);
    return factory.newConnection();
  }
}

Save the publisher as src/main/java/Publisher.java:

import static java.nio.charset.StandardCharsets.UTF_8;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;

public class Publisher {
  public static void main(String[] args) throws Exception {
    try (Connection connection = Amqp.connect()) {
      publish(connection.createChannel(), "hello", "Hello CloudAMQP!");
    }
  }

  public static void publish(Channel channel, String queue, String message) throws Exception {
    channel.queueDeclare(queue, true, false, false, null);
    channel.confirmSelect();
    channel.basicPublish("", queue, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(UTF_8));
    channel.waitForConfirmsOrDie(5_000);
    System.out.println(" [x] Sent '" + message + "'");
  }
}

Save the consumer as src/main/java/Worker.java:

import static java.nio.charset.StandardCharsets.UTF_8;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.function.Consumer;

public class Worker {
  public static void main(String[] args) throws Exception {
    Connection connection = Amqp.connect();
    consume(connection.createChannel(), "hello", message -> System.out.println(" [x] Received '" + message + "'"));
    System.out.println(" [*] Waiting for messages");
  }

  public static void consume(Channel channel, String queue, Consumer<String> handler) throws IOException {
    channel.queueDeclare(queue, true, false, false, null);
    channel.basicQos(10);
    channel.basicConsume(queue, false, (consumerTag, delivery) -> {
      handler.accept(new String(delivery.getBody(), UTF_8));
      channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }, consumerTag -> {});
  }
}

Run mvn compile exec:java -Dexec.mainClass=Worker in one terminal and mvn compile exec:java -Dexec.mainClass=Publisher in another.

A runnable example is in java/rabbitmq-client in cloudamqp/examples.

Good to know

Connect with the URL. factory.setUri(url), where the URL is the one on your instance’s details page in the CloudAMQP console. With amqps:// the server certificate and hostname are verified against the JVM’s trust store.
Automatic recovery is on by default: after a network failure the client reconnects and restores channels, queues and consumers.
Use publisher confirms (confirmSelect, waitForConfirmsOrDie) and persistent messages, so a message isn’t lost if the connection or broker goes down.
Ack after processing, and set a prefetch limit with basicQos, so a consumer doesn’t buffer more messages than it can handle.

See the Java client API guide for more.

Other Java libraries

Spring AMQP, for Spring applications. Set spring.rabbitmq.addresses to your CloudAMQP URL.

← PHP

Clojure →