import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TransactionalMessageProducer {

    public static void main(String[] args) throws Exception {
        // 创建事务消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("TransactionProducerGroup");
        producer.setNamesrvAddr("localhost:9876");

        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // 执行本地事务
                try {
                    // 模拟本地事务执行
                    boolean isSuccess = executeLocalBusiness();
                    if (isSuccess) {
                        // 本地事务成功，提交消息
                        return LocalTransactionState.COMMIT_MESSAGE;
                    } else {
                        // 本地事务失败，回滚消息
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                } catch (Exception e) {
                    // 本地事务执行异常，返回未知状态，等待回查
                    return LocalTransactionState.UNKNOW;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // 处理 Half 消息未处理超时的情况
                System.out.println("Half 消息未处理超时，执行回查逻辑...");

                // 检查本地事务状态
                boolean isLocalTransactionSuccess = checkLocalTransactionStatus(msg);
                if (isLocalTransactionSuccess) {
                    // 本地事务已成功，提交消息
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else {
                    // 本地事务失败，回滚消息
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
        });

        // 启动生产者
        producer.start();

        // 创建消息
        Message msg = new Message("TransactionTopic", "TagA", "Hello RocketMQ Transaction".getBytes());

        // 发送事务消息
        producer.sendMessageInTransaction(msg, null);

        // 关闭生产者
        producer.shutdown();
    }

    private static boolean executeLocalBusiness() {
        // 模拟本地事务执行逻辑
        // 这里可以根据业务需求执行数据库操作或其他业务逻辑
        // 返回 true 表示本地事务成功，false 表示失败
        return false; // 模拟本地事务失败
    }

    private static boolean checkLocalTransactionStatus(MessageExt msg) {
        // 模拟检查本地事务状态的逻辑
        // 这里可以查询数据库或调用其他服务，确认本地事务是否成功
        // 返回 true 表示本地事务成功，false 表示失败
        return true; // 模拟本地事务已成功
    }
}

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;

public class OrderedProducer {

    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("OrderedProducerGroup");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 模拟订单状态变更消息
        String[] orderStatus = {"创建订单", "支付订单", "发货订单", "完成订单"};
        for (int orderId = 1; orderId <= 10; orderId++) {
            for (int statusIndex = 0; statusIndex < orderStatus.length; statusIndex++) {
                // 创建消息
                String messageBody = "订单ID: " + orderId + ", 状态: " + orderStatus[statusIndex];
                Message msg = new Message("OrderTopic", "TagA", messageBody.getBytes());

                // 发送顺序消息
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        // 根据订单ID选择队列，确保同一订单的消息发送到同一个队列
                        int orderId = (int) arg;
                        int index = orderId % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId); // 将订单ID作为参数传递

                System.out.println("发送消息: " + messageBody + ", 发送结果: " + sendResult);
            }
        }

        // 关闭生产者
        producer.shutdown();
    }
}