消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
- 文档首页
消息队列 RocketMQ版SDK 参考RocketMQ 5.xJava 5.x SDK事务消息
复制全文
下载 pdf事务消息
文档反馈
问问助手本文提供使用 Java SDK 收发事务消息的示例代码供您参考。
发送事务消息
发送事务消息需要在控制台创建事务消息类型的 Topic，在发送 half 消息后，用户可以根据自身业务逻辑对消息选择：
- TransactionResolution.COMMIT：提交事务。事务提交之后，之前暂存的消息对消费者可见。
- TransactionResolution.ROLLBACK：回滚事务。事务回滚，之前暂存的消息会被标记为已经回滚。
- TransactionResolution.UNKNOWN：无法判断事务状态。业务逻辑暂时无法判断是否需要提交之前暂存的消息状态，希望 RocketMQ 服务端可以稍后触发回调的逻辑进行重试。
此外，事务消息需要自定义checker，以便服务端在未收到确认时触发回查。
public class RocketMQTransactionProducer {
private static final Logger log = LoggerFactory.getLogger(RocketMQTransactionProducer.class);
private RocketMQTransactionProducer() {
}
public static void main(String[] args) throws ClientException, IOException {
final ClientServiceProvider provider = ClientServiceProvider.loadService();
// 设置为您从火山引擎消息队列 RocketMQ版控制台获取的接入点信息，类似“{INSTANCE_ID}.rocketmq.ivolces.com:8080”。
String endpoints = "rocketmq-xxx.rocketmq.ivolces.com:8080";
// 根据实例支持的认证方式，填写对应的身份认证信息：
// 密钥管理：accessKey 为 AccessKey ID，secretKey 为 AccessKey Secret。
// 权限管理：accessKey 为 ACL 用户名，secretKey 为 ACL 密码。
String accessKey = "xxxx";
String secretKey = "xxxx";
SessionCredentialsProvider sessionCredentialsProvider =
new StaticSessionCredentialsProvider(accessKey, secretKey);
ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
.setEndpoints(endpoints)
// 暂不支持 SSL 模式。
.enableSsl(false)
.setCredentialProvider(sessionCredentialsProvider)
.build();
// 定义消息体。
String topic = "topic-transaction"; // 已在控制台创建的事务消息类型 Topic。
final Producer producer = provider.newProducerBuilder()
.setClientConfiguration(clientConfiguration)
.setTopics(topic)
// 定义 checker。
.setTransactionChecker(messageView -> {
log.info("Receive transactional message check, message={}", messageView);
// Return the transaction resolution according to your business logic.
return TransactionResolution.COMMIT;
})
.build();
final Transaction transaction = producer.beginTransaction();
byte[] body = "This is a transaction message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
String tag = "*";
final Message message = provider.newMessageBuilder()
// Set topic for the current message.
.setTopic(topic)
// Message secondary classifier of message besides topic.
.setTag(tag)
// Key(s) of the message, another way to mark message besides message id.
.setKeys("yourMessageKey-1ff69ada8e0e")
.setBody(body)
.build();
try {
final SendReceipt sendReceipt = producer.send(message, transaction);
log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (Throwable t) {
log.error("Failed to send message", t);
}
// Commit the transaction.
transaction.commit();
// Or rollback the transaction.
// transaction.rollback();
// Close the producer when you don't need it anymore.
// You could close it manually or add this into the JVM shutdown hook.
producer.close();
}
}
订阅事务消息
事务消息的订阅方式与普通消息一致，示例代码如下所示。
public class RocketMQTransactionConsumer {
private static final Logger log = LoggerFactory.getLogger(RocketMQTransactionConsumer.class);
private RocketMQTransactionConsumer() {
}
public static void main(String[] args) throws ClientException, InterruptedException, IOException {
final ClientServiceProvider provider = ClientServiceProvider.loadService();
// 设置为您从火山引擎消息队列 RocketMQ版控制台获取的接入点信息，类似“{INSTANCE_ID}.rocketmq.ivolces.com:8080”。
String endpoints = "rocketmq-xxx.rocketmq.ivolces.com:8080";
// 根据实例支持的认证方式，填写对应的身份认证信息：
// 密钥管理：accessKey 为 AccessKey ID，secretKey 为 AccessKey Secret。
// 权限管理：accessKey 为 ACL 用户名，secretKey 为 ACL 密码。
String accessKey = "xxxx";
String secretKey = "xxxx";
SessionCredentialsProvider sessionCredentialsProvider =
new StaticSessionCredentialsProvider(accessKey, secretKey);
ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
.setEndpoints(endpoints)
// 暂不支持 SSL 模式。
.enableSsl(false)
.setCredentialProvider(sessionCredentialsProvider)
.build();
String tag = "yourMessageTagA";
FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
String consumerGroup = "yourConsumerGroup";
String topic = "yourTopic";
// In most case, you don't need to create too many consumers, singleton pattern is recommended.
PushConsumer pushConsumer = provider.newPushConsumerBuilder()
.setClientConfiguration(clientConfiguration)
// Set the consumer group name.
.setConsumerGroup(consumerGroup)
// Set the subscription for the consumer.
.setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
.setMessageListener(messageView -> {
// Handle the received message and return consume result.
log.info("Consume message={}", messageView);
return ConsumeResult.SUCCESS;
})
.build();
// Block the main thread, no need for production environment.
Thread.sleep(Long.MAX_VALUE);
// Close the push consumer when you don't need it anymore.
// You could close it manually or add this into the JVM shutdown hook.
pushConsumer.close();
}
}
最近更新时间：2026.08.26 15:19:56
这个页面对您有帮助吗？
有用
有用
无用
无用