This document describes how to use open-source SDK to send and receive messages using the SDK for Java as an example and helps you better understand the message sending and receiving processes.
NoteThe following takes the Java client as an example. For clients in other languages, see TDMQ for RabbitMQ.
Add the following dependencies to the pom.xml
file:
<!-- in your <dependencies> block -->
<dependency>
<groupid>com.rabbitmq</groupid>
<artifactid>amqp-client</artifactid>
<version>5.13.0</version>
</dependency>
Compile and run MessageProducer.java
.
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.tencent.tdmq.demo.cloud.Constant;
/**
* Message producer
*/
public class MessageProducer {
/**
* Exchange name
*/
private static final String EXCHANGE_NAME = "exchange_name";
public static void main(String[] args) throws Exception {
// Connection factory
ConnectionFactory factory = new ConnectionFactory();
// Set the service address (replace with the access point address copied in the console)
factory.setUri("amqp://***");
// Set vhost (replace with the vhost name copied in the console)
factory.setVirtualHost(VHOST_NAME);
// Set the username (use the role name in the permission configuration of the vhost)
factory.setUsername(USERNAME);
// Set the password (use the role key)
factory.setPassword("eyJh****");
// Get the connection address and establish the channel
try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
// Bind the message exchange (`EXCHANGE_NAME` must exist in the TDMQ for RabbitMQ console, and the exchange type must be the same as that in the console)
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
for (int i = 0; i < 10; i++) {
String message = "this is rabbitmq message " + i;
// Publish a message to the exchange, which will automatically deliver the message to the corresponding queue
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println(" [producer] Sent '" + message + "'");
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
Parameter | Description |
---|---|
EXCHANGE_NAME | Exchange name, which can be obtained from the exchange list in the console. |
factory.setUri | Cluster access address, which can be obtained in the console by clicking Get Access Address in the Operation column on the Cluster page. |
factory.setVirtualHost | Vhost name in the format of "cluster ID + | + vhost name". |
factory.setUsername | Role name, which can be copied on the Role Management page. |
factory.setPassword | Role token, which can be copied in the Token column on the Role Management page. |
Compile and run MessageConsumer.java
.
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.tencent.tdmq.demo.cloud.Constant;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
/**
* Message consumer
*/
public class MessageConsumer1 {
/**
* Queue name
*/
public static final String QUEUE_NAME = "queue_name";
/**
* Exchange name
*/
private static final String EXCHANGE_NAME = "exchange_name";
public static void main(String[] args) throws Exception {
// Connection factory
ConnectionFactory factory = new ConnectionFactory();
// Set the service address (replace with the access point address copied in the console)
factory.setUri("amqp://***");
// Set vhost (replace with the vhost name copied in the console)
factory.setVirtualHost(VHOST_NAME);
// Set the username (use the role name in the permission configuration of the vhost)
factory.setUsername(USERNAME);
// Set the password (use the role key)
factory.setPassword("eyJh****");
// Get the connection address
Connection connection = factory.newConnection();
// Establish a channel
Channel channel = connection.createChannel();
// Bind the message exchange
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// Declare the queue message
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// Bind the message exchange (`EXCHANGE_NAME` must exist in the TDMQ for RabbitMQ console, and the exchange type must be the same as that in the console)
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
System.out.println(" [Consumer1] Waiting for messages.");
// Subscribe to the message
channel.basicConsume(QUEUE_NAME, false, "ConsumerTag", new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
// Received message for business logic processing
System.out.println("Received: " + new String(body, StandardCharsets.UTF_8) + ", deliveryTag: " + envelope.getDeliveryTag() + ", messageId: " + properties.getMessageId());
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}
Parameter | Description |
---|---|
QUEUE_NAME | Queue name, which can be obtained from the queue list in the console. |
EXCHANGE_NAME | Exchange name, which can be obtained from the exchange list in the console. |
factory.setUri | Cluster access address, which can be obtained in the console by clicking Get Access Address in the Operation column on the Cluster page. |
factory.setVirtualHost | Vhost name in the format of "cluster ID + | + vhost name". |
factory.setUsername | Role name, which can be copied on the Role Management page. |
factory.setPassword | Role token, which can be copied in the Token column on the Role Management page. |
To check whether messages are sent to TDMQ for RabbitMQ successfully, view the connected consumer status on the Cluster > Queue page in the console.
Note:Above is a sample based on the pub/sub pattern of RabbitMQ. For more samples, see Demo or RabbitMQ Tutorials.
Was this page helpful?