Cloud Native Message Engine
Cloud Native Message Engine
Document
- Documentation
Cloud Native Message EngineBMQ Instance User GuideSDK referenceJava SDK
Copy page
Download PDFSDK reference
Java SDK
Copy pageDownload PDFJava SDK
This topic describes how to use the Java SDK to connect to the cloud-native messaging engine BMQ and send and receive messages.
Prerequisites
- Create a resource instance and obtain the endpoint address. For details, see Create a BMQ instance.
- You should install the Java environment in advance, including JDK version 1.8 or later and Maven version 3.5 or later. For detailed instructions, see Install JDK and Install Maven.
- You need to open port 9092 in advance in the security group to which the instance belongs. For detailed instructions, see Add security group access rules.
- (Optional) To authenticate using a SASL username and password, you must create a user and obtain the password in advance. For detailed instructions, see Create SASL user.
Install Java dependencies
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.2.0</version>
</dependency>
Send messages
Write and run BmqProducerDemo.java to send messages.
PLAINTEXT
When connecting to the BMQ instance using the endpoint address with the PLAINTEXT protocol, authentication is not required.
//View the corresponding access point information in the console
String server = "xxx.";
//Topic of the message applied for in the console
String topic = "this is your topic.";
//Test message content
String value = "this is test message value.";
//Number of messages to send
int count = 100;
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
try {
for (int i = 0; i < count; i++) {
RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, value + i++))
.get(5, TimeUnit.SECONDS);
logger.info("recordMetadata topic={}, partition={}, offset={}, count = {}.",
recordMetadata.topic(),
recordMetadata.partition(),
recordMetadata.offset(),
i);
}
} catch (Throwable e) {
logger.error("produce error", e);
}
producer.flush();
producer.close();
SASL_PLAINTEXT
Authenticate using a SASL username and password. You need to obtain the username and password of an authorized user. For details on how to obtain them, see View SASL user password.
//View the corresponding access point information in the console
String server = "xxx.";
//Topic of the message applied for in the console
String topic = "this is your topic.";
//Test message content
String value = "this is test message value.";
//Number of messages to send
int count = 100;
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//SASL_PLAINTEXT configuration
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
properties.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"User name\" password=\"Password\";");
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
try {
for (int i = 0; i < count; i++) {
RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, value + i++))
.get(5, TimeUnit.SECONDS);
logger.info("recordMetadata topic={}, partition={}, offset={}, count = {}.",
recordMetadata.topic(),
recordMetadata.partition(),
recordMetadata.offset(),
i);
}
} catch (Throwable e) {
logger.error("produce error", e);
}
producer.flush();
producer.close();
SASL_SSL
Authenticate using a SASL username and password. You need to obtain the username and password of an authorized user. For details on how to obtain them, see View SASL user password.
//View the corresponding access point information in the console
String server = "xxx.";
//Topic of the message applied for in the console
String topic = "this is your topic.";
//Test message content
String value = "this is test message value.";
//Number of messages to send
int count = 100;
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//SASL_SSL configuration
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
properties.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"User name\" password=\"Password\";");
properties.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
try {
for (int i = 0; i < count; i++) {
RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, value + i++))
.get(5, TimeUnit.SECONDS);
logger.info("recordMetadata topic={}, partition={}, offset={}, count = {}.",
recordMetadata.topic(),
recordMetadata.partition(),
recordMetadata.offset(),
i);
}
} catch (Throwable e) {
logger.error("produce error", e);
}
producer.flush();
producer.close();
consume messages
Write and run BmqConsumerDemo.java to consume messages.
PLAINTEXT
When connecting to the BMQ instance using the endpoint address with the PLAINTEXT protocol, authentication is not required.
//View the corresponding endpoint information in the console
String server = "xxx.";
//The Topic to which the message belongs, as applied for in the console
String topic = "this is your topic.";
//The consumerGroup for consuming messages, as applied for in the console
String group = "this is your group.";
//Offset consumption policy: earliest, latest, none
String offsetReset = "earliest";
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, group);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetReset);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
List<String> topicList = Lists.newArrayList(topic);
consumer.subscribe(topicList);
while (true) {
try {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> record : records) {
logger.info("consumed record, topic={}, partition={}, offset={}, key={}, value={}",
record.topic(),
record.partition(),
record.offset(),
record.key(),
record.value());
}
} catch (Exception e) {
logger.error("consume error", e);
}
}
SASL_PLAINTEXT
Authenticate using a SASL username and password. You need to obtain the username and password of an authorized user. For details on how to obtain them, see View SASL user password.
//View the corresponding access point information in the console
String server = "xxx.";
//Topic of the message applied for in the console
String topic = "this is your topic.";
//Consumer group applied for in the console for consuming messages
String group = "this is your group.";
//Offset policy: earliest, latest, none
String offsetReset = "earliest";
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, group);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetReset);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//SASL_PLAINTEXT configuration
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
properties.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"User name\" password=\"Password\";");
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
List<String> topicList = Lists.newArrayList(topic);
consumer.subscribe(topicList);
while (true) {
try {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> record : records) {
logger.info("consumed record, topic={}, partition={}, offset={}, key={}, value={}",
record.topic(),
record.partition(),
record.offset(),
record.key(),
record.value());
}
} catch (Exception e) {
logger.error("consume error", e);
}
}
SASL_SSL
Authenticate using a SASL username and password. You need to obtain the username and password of an authorized user. For details on how to obtain them, see View SASL user password.
//View the corresponding endpoint information in the console
String server = "xxx.";
//The Topic to which the message belongs, as applied for in the console
String topic = "this is your topic.";
//The consumerGroup for consuming messages, as applied for in the console
String group = "this is your group.";
//Offset consumption policy: earliest, latest, none
String offsetReset = "earliest";
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, group);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetReset);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//Set the maximum amount of data to pull; must be less than the public network bandwidth
properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 512000);
properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 512000);
//SASL_SSL configuration
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
properties.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"User name\" password=\"Password\";");
properties.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
List<String> topicList = Lists.newArrayList(topic);
consumer.subscribe(topicList);
while (true) {
try {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> record : records) {
logger.info("consumed record, topic={}, partition={}, offset={}, key={}, value={}",
record.topic(),
record.partition(),
record.offset(),
record.key(),
record.value());
}
} catch (Exception e) {
logger.error("consume error", e);
}
}
Last updated: 2026.01.26 15:46:58
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
