in ons-sample/src/main/java/org/apache/rocketmq/ons/sample/producer/SimpleTransactionProducer.java [33:62]
public static void main(String[] args) {
Properties tranProducerProperties = new Properties();
tranProducerProperties.setProperty(PropertyKeyConst.GROUP_ID, MQConfig.GROUP_ID);
tranProducerProperties.setProperty(PropertyKeyConst.AccessKey, MQConfig.ACCESS_KEY);
tranProducerProperties.setProperty(PropertyKeyConst.SecretKey, MQConfig.SECRET_KEY);
tranProducerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR, MQConfig.NAMESRV_ADDR);
LocalTransactionCheckerImpl localTransactionChecker = new LocalTransactionCheckerImpl();
TransactionProducer transactionProducer = ONSFactory.createTransactionProducer(tranProducerProperties, localTransactionChecker);
transactionProducer.start();
Message message = new Message(MQConfig.TOPIC, MQConfig.TAG, "MQ send transaction message test".getBytes());
for (int i = 0; i < 10; i++) {
try {
SendResult sendResult = transactionProducer.send(message, new LocalTransactionExecuter() {
@Override
public TransactionStatus execute(Message msg, Object arg) {
System.out.printf("Execute local transaction and return TransactionStatus. %n");
return TransactionStatus.CommitTransaction;
}
}, null);
assert sendResult != null;
} catch (ONSClientException e) {
System.out.printf(new Date() + " Send mq message failed! Topic is: %s%n", MQConfig.TOPIC);
e.printStackTrace();
}
}
System.out.printf("Send transaction message success. %n");
}