云原生消息引擎
云原生消息引擎
云原生消息引擎
文档指南
请输入
产品动态
BMQ 实例用户指南
控制台指南
实例管理
SDK 参考
API 参考
实例管理
实例连接管理
Topic 管理
MQTT 实例用户指南
控制台指南
MQTT Agent
API 参考
MQTT密钥
Dashboard指南
访问控制
- 文档首页
云原生消息引擎BMQ 实例用户指南快速入门使用 Java SDK 收发消息
复制全文
下载 pdf使用 Java SDK 收发消息
文档反馈
问问助手本文以调用 Java SDK 为例，介绍如何通过 SDK 接入 BMQ 实例并收发消息。
准备工作
- 您需要提前在实例所属安全组中放开 9092 端口。具体操作，请参见添加安全组访问规则。
- （可选）您如果需要通过 SASL 用户名和密码进行鉴权，还需提前创建用户并获取密码。具体操作，请参见创建 SASL 用户。
添加依赖
在 pom.xml 文件中添加以下依赖。
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.2.0</version>
</dependency>
发送消息
编写并运行BmqProducerDemo.java发送消息。
//在控制台查看对应接入点信息
String server = "xxx.";
//在控制台申请的消息所属Topic
String topic = "this is your topic.";
//测试消息内容
String value = "this is test message value.";
//发送消息条数
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();
消费消息
编写并运行 BmqConsumerDemo.java 消费消息。
//在控制台查看对应接入点信息
String server = "xxx.";
//在控制台申请的消息所属Topic
String topic = "this is your topic.";
//在控制台申请消费消息的consumerGroup
String group = "this is your group.";
//消费offset策略：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);
}
}
最近更新时间：2025.08.04 17:30:00
这个页面对您有帮助吗？
有用
有用
无用
无用