TransProducer.java

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQTransProducer;
import com.aliyun.mq.http.common.AckMessageException;
import com.aliyun.mq.http.model.Message;
import com.aliyun.mq.http.model.TopicMessage;

import java.util.List;

public class TransProducer {


    static void processCommitRollError(Throwable e) {
        if (e instanceof AckMessageException) {
            AckMessageException errors = (AckMessageException) e;
            System.out.println("Commit/Roll transaction error, requestId is:" + errors.getRequestId() + ", fail handles:");
            if (errors.getErrorMessages() != null) {
                for (String errorHandle :errors.getErrorMessages().keySet()) {
                    System.out.println("Handle:" + errorHandle + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                            + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                }
            }
        }
    }

    public static void main(String[] args) throws Throwable {
        MQClient mqClient = new MQClient(
                // 设置HTTP接入域名（此处以公共云生产环境为例）
                "${HTTP_ENDPOINT}",
                // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
                "${ACCESS_KEY}",
                // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
                "${SECRET_KEY}"
        );

        // 所属的 Topic
        final String topic = "${TOPIC}";
        // Topic所属实例ID，默认实例为空
        final String instanceId = "${INSTANCE_ID}";
        // 您在控制台创建的 Consumer ID(Group ID)
        final String groupId = "${GROUP_ID}";

        final MQTransProducer mqTransProducer = mqClient.getTransProducer(instanceId, topic, groupId);

        for (int i = 0; i < 4; i++) {
            TopicMessage topicMessage = new TopicMessage();
            topicMessage.setMessageBody("trans_msg");
            topicMessage.setMessageTag("a");
            topicMessage.setMessageKey(String.valueOf(System.currentTimeMillis()));
            // 设置事务第一次回查的时间，为相对时间，单位：秒，范围为10~300s之间
            // 第一次事务回查后如果消息没有commit或者rollback，则之后每隔10s左右会回查一次，总共回查一天
            topicMessage.setTransCheckImmunityTime(10);
            topicMessage.getProperties().put("a", String.valueOf(i));

            TopicMessage pubResultMsg = null;
            pubResultMsg = mqTransProducer.publishMessage(topicMessage);
            System.out.println("Send---->msgId is: " + pubResultMsg.getMessageId()
                    + ", bodyMD5 is: " + pubResultMsg.getMessageBodyMD5()
                    + ", Handle: " + pubResultMsg.getReceiptHandle()
            );
            if (pubResultMsg != null && pubResultMsg.getReceiptHandle() != null) {
                if (i == 0) {
                    // 发送完事务消息后能获取到半消息句柄，可以直接commit/rollback事务消息
                    try {
                        mqTransProducer.commit(pubResultMsg.getReceiptHandle());
                        System.out.println(String.format("MessageId:%s, commit", pubResultMsg.getMessageId()));
                    } catch (Throwable e) {
                        // 如果Commit/Rollback时超过了TransCheckImmunityTime则会失败
                        if (e instanceof AckMessageException) {
                            processCommitRollError(e);
                            continue;
                        }
                    }
                }
            }
        }

        // 客户端需要有一个线程或者进程来消费没有确认的事务消息
        // 示例这里启动一个线程来检查没有确认的事务消息
        Thread t = new Thread(new Runnable() {
            public void run() {
                int count = 0;
                while(true) {
                    try {
                        if (count == 3) {
                            break;
                        }
                        List<Message> messages = mqTransProducer.consumeHalfMessage(3, 3);
                        if (messages == null) {
                            System.out.println("No Half message!");
                            continue;
                        }
                        System.out.println(String.format("Half---->MessageId:%s,Properties:%s,Body:%s,Latency:%d",
                                messages.get(0).getMessageId(),
                                messages.get(0).getProperties(),
                                messages.get(0).getMessageBodyString(),
                                System.currentTimeMillis() - messages.get(0).getPublishTime()));

                        for (Message message : messages) {
                            try {
                                if (Integer.valueOf(message.getProperties().get("a")) == 1) {
                                    // 确认提交事务消息
                                    mqTransProducer.commit(message.getReceiptHandle());
                                    count++;
                                    System.out.println(String.format("MessageId:%s, commit", message.getMessageId()));
                                } else if (Integer.valueOf(message.getProperties().get("a")) == 2
                                        && message.getConsumedTimes() > 1) {
                                    // 确认提交事务消息
                                    mqTransProducer.commit(message.getReceiptHandle());
                                    count++;
                                    System.out.println(String.format("MessageId:%s, commit", message.getMessageId()));
                                } else if (Integer.valueOf(message.getProperties().get("a")) == 3) {
                                    // 确认回滚事务消息
                                    mqTransProducer.rollback(message.getReceiptHandle());
                                    count++;
                                    System.out.println(String.format("MessageId:%s, rollback", message.getMessageId()));
                                } else {
                                    // 什么都不做，下次再检查
                                    System.out.println(String.format("MessageId:%s, unknown", message.getMessageId()));
                                }
                            } catch (Throwable e) {
                                // 如果Commit/Rollback时超过了TransCheckImmunityTime（针对发送事务消息的句柄）或者超过10s（针对consumeHalfMessage的句柄）则会失败
                                processCommitRollError(e);
                            }
                        }
                    } catch (Throwable e) {
                        System.out.println(e.getMessage());
                    }
                }
            }
        });

        t.start();

        t.join();

        mqClient.close();
    }

}