From f60028e276cf690412a81a21dfe458bec49487f6 Mon Sep 17 00:00:00 2001 From: javahongxi Date: Fri, 26 Oct 2018 15:42:49 +0800 Subject: [PATCH] transaction --- .../transaction/TransactionListenerImpl.java | 57 +++++++++++++++ .../transaction/TransactionProducer.java | 70 +++++++++++++++++++ 2 files changed, 127 insertions(+) create mode 100644 whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/mq/rocketmq/transaction/TransactionListenerImpl.java create mode 100644 whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/mq/rocketmq/transaction/TransactionProducer.java diff --git a/whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/mq/rocketmq/transaction/TransactionListenerImpl.java b/whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/mq/rocketmq/transaction/TransactionListenerImpl.java new file mode 100644 index 00000000..6ef51797 --- /dev/null +++ b/whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/mq/rocketmq/transaction/TransactionListenerImpl.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.hongxi.whatsmars.mq.rocketmq.transaction; + +import org.apache.rocketmq.client.producer.LocalTransactionState; +import org.apache.rocketmq.client.producer.TransactionListener; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageExt; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +public class TransactionListenerImpl implements TransactionListener { + private AtomicInteger transactionIndex = new AtomicInteger(0); + + private ConcurrentHashMap localTrans = new ConcurrentHashMap<>(); + + @Override + public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { + int value = transactionIndex.getAndIncrement(); + int status = value % 3; + localTrans.put(msg.getTransactionId(), status); + return LocalTransactionState.UNKNOW; + } + + @Override + public LocalTransactionState checkLocalTransaction(MessageExt msg) { + Integer status = localTrans.get(msg.getTransactionId()); + if (null != status) { + switch (status) { + case 0: + return LocalTransactionState.UNKNOW; + case 1: + return LocalTransactionState.COMMIT_MESSAGE; + case 2: + return LocalTransactionState.ROLLBACK_MESSAGE; + default: + return LocalTransactionState.COMMIT_MESSAGE; + } + } + return LocalTransactionState.COMMIT_MESSAGE; + } +} \ No newline at end of file diff --git a/whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/mq/rocketmq/transaction/TransactionProducer.java b/whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/mq/rocketmq/transaction/TransactionProducer.java new file mode 100644 index 00000000..15243e01 --- /dev/null +++ b/whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/mq/rocketmq/transaction/TransactionProducer.java @@ -0,0 +1,70 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.hongxi.whatsmars.mq.rocketmq.transaction; + +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.producer.SendResult; +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.remoting.common.RemotingHelper; + +import java.io.UnsupportedEncodingException; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +public class TransactionProducer { + public static void main(String[] args) throws MQClientException, InterruptedException { + TransactionListener transactionListener = new TransactionListenerImpl(); + TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name"); + ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue(2000), new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r); + thread.setName("client-transaction-msg-check-thread"); + return thread; + } + }); + + producer.setExecutorService(executorService); + producer.setTransactionListener(transactionListener); + producer.start(); + + String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"}; + for (int i = 0; i < 10; i++) { + try { + Message msg = + new Message("TopicTest1234", tags[i % tags.length], "KEY" + i, + ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); + SendResult sendResult = producer.sendMessageInTransaction(msg, null); + System.out.printf("%s%n", sendResult); + + Thread.sleep(10); + } catch (MQClientException | UnsupportedEncodingException e) { + e.printStackTrace(); + } + } + + for (int i = 0; i < 100000; i++) { + Thread.sleep(1000); + } + producer.shutdown(); + } +} \ No newline at end of file -- GitLab