transaction msg

Author: lazycece

springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/run/Runner.java

@@ -0,0 +1,29 @@
+package com.lazycece.sbac.rocketmq.run;
+
+import com.lazycece.sbac.rocketmq.simple.RocketmqProducer;
+import com.lazycece.sbac.rocketmq.transaction.TransactionProducer;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+
+/**
+ * @author lazycece
+ * @date 2019/8/21
+ */
+@Component
+public class Runner implements CommandLineRunner {
+
+ @Resource
+ private RocketmqProducer rocketmqProducer;
+ @Resource
+ private TransactionProducer transactionProducer;
+
+ @Override
+ public void run(String... args) throws Exception {
+// rocketmqProducer.sync();
+ transactionProducer.produce();
+
+ }
+}

…/rocketmq/consumer/RocketmqConsumer.java …ac/rocketmq/simple/RocketmqConsumer.javaspringboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/consumer/RocketmqConsumer.java renamed to springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/simple/RocketmqConsumer.java springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/consumer/RocketmqConsumer.java renamed to springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/simple/RocketmqConsumer.java

@@ -1,20 +1,20 @@
-package com.lazycece.sbac.rocketmq.consumer;
+package com.lazycece.sbac.rocketmq.simple;
 
 import com.lazycece.sbac.rocketmq.message.Message;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
-import org.springframework.stereotype.Service;
+import org.springframework.stereotype.Component;
 
 /**
  * @author lazycece
  * @date 2019/6/6
  */
 @Slf4j
-@Service
+@Component
 public class RocketmqConsumer {
 
- @Service
+ @Component
  @RocketMQMessageListener(topic = "topic-queue-one", consumerGroup = "consumer_topic-queue-one")
  public class ConsumerOne implements RocketMQListener<Message> {
  @Override
@@ -23,7 +23,7 @@ public void onMessage(Message message) {
  }
  }
 
- @Service
+ @Component
  @RocketMQMessageListener(topic = "topic-queue-two", consumerGroup = "consumer_topic-queue-two")
  public class ConsumerTwo implements RocketMQListener<String> {
  @Override

…/rocketmq/producer/RocketmqProducer.java …ac/rocketmq/simple/RocketmqProducer.javaspringboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/producer/RocketmqProducer.java renamed to springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/simple/RocketmqProducer.java springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/producer/RocketmqProducer.java renamed to springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/simple/RocketmqProducer.java

@@ -1,33 +1,43 @@
-package com.lazycece.sbac.rocketmq.producer;
+package com.lazycece.sbac.rocketmq.simple;
 
 import com.lazycece.sbac.rocketmq.message.Message;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
 import java.util.UUID;
 
+
 /**
  * @author lazycece
  * @date 2019/6/6
  */
 @Component
-public class RocketmqProducer implements CommandLineRunner {
+public class RocketmqProducer {
 
+ @Resource
  private RocketMQTemplate rocketMQTemplate;
 
- @Autowired
- public RocketmqProducer(RocketMQTemplate rocketMQTemplate) {
- this.rocketMQTemplate = rocketMQTemplate;
- }
-
- @Override
- public void run(String... args) throws Exception {
+ /**
+ * 同步发送
+ */
+ public void sync() {
  Message<String> message = new Message<>();
  message.setId(UUID.randomUUID().toString());
  message.setContent("Hello, springboot-ac-rocketmq !");
  rocketMQTemplate.convertAndSend("topic-queue-one", message);
  rocketMQTemplate.convertAndSend("topic-queue-two", "Hello, springboot-ac-rocketmq !");
  }
+
+ /**
+ * 异步发送
+ */
+ public void async() {
+ Message<String> message = new Message<>();
+ message.setId(UUID.randomUUID().toString());
+ message.setContent("send async message!");
+// rocketMQTemplate.
+ }
+
+
 }

springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/transaction/TransactionListenerImpl.java

@@ -6,7 +6,6 @@
 import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
 import org.springframework.messaging.Message;
 
-
 /**
  * @author lazycece
  * @date 2019/8/20
@@ -18,6 +17,7 @@ public class TransactionListenerImpl implements RocketMQLocalTransactionListener
  @Override
  public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
 
+ // 模拟本地事务不通过
  log.info("============== executeLocalTransaction");
 
  return RocketMQLocalTransactionState.UNKNOWN;
@@ -26,8 +26,11 @@ public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object
  @Override
  public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
 
+ // 模拟回查本地事务
  log.info("============== checkLocalTransaction");
 
+ // todo 收不到回查消息
+
  return RocketMQLocalTransactionState.COMMIT;
  }
 }

springboot-ac-rocketmq/src/main/java/com/lazycece/sbac/rocketmq/transaction/TransactionProducer.java

@@ -1,7 +1,10 @@
 package com.lazycece.sbac.rocketmq.transaction;
 
 import com.lazycece.sbac.rocketmq.message.Message;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 import java.util.UUID;
@@ -10,16 +13,20 @@
  * @author lazycece
  * @date 2019/8/20
  */
+@Component
+@Slf4j
 public class TransactionProducer {
 
  @Resource
  private RocketMQTemplate rocketMQTemplate;
 
- public void produce(){
+ public void produce() {
  Message<String> message = new Message<>();
  message.setId(UUID.randomUUID().toString());
  message.setContent("transaction message");
-// rocketMQTemplate.sendMessageInTransaction("tx-group","topic-tx",message,null);
+ log.info("========sending message=========");
+ rocketMQTemplate.sendMessageInTransaction("tx-group", "topic-tx", MessageBuilder.withPayload(message).build(), null);
+ log.info("========finish send =========");
  }
 
 }

springboot-ac-rocketmq/src/main/resources/application.yml

@@ -1,5 +1,4 @@
 rocketmq:
-# name-server: 127.0.0.1:9876
- name-server: 10.40.7.18:9876
+ name-server: 127.0.0.1:9876
  producer:
- group: rocketmq-group
+ group: rocketmq-group