rcoketmq集中管理

admin

2020-07-01 c54fb6a88876be994906d57d2d18e844686964d0

 fanli/src/main/java/com/yeshi/fanli/service/manger/msg/RocketMQManager.java

@@ -6,6 +6,7 @@


import com.aliyun.openservices.ons.api.Message;

import com.aliyun.openservices.ons.api.Producer;

import com.aliyun.openservices.ons.api.SendResult;

import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;

import com.aliyun.openservices.ons.api.transaction.TransactionProducer;

import com.aliyun.openservices.ons.api.transaction.TransactionStatus;

@@ -30,8 +31,8 @@
    * void 返回类型

    * @throws

    */

   public void sendNormalMsg(Message message, String key) {

      sendNormalMsg(message, null, key);

   public SendResult sendNormalMsg(Message message, String key) {

      return sendNormalMsg(message, null, key);

   }



   /**

@@ -43,14 +44,14 @@
    * void 返回类型

    * @throws

    */

   public void sendNormalMsg(Message message, Long delayTimeMS, String key) {

   public SendResult sendNormalMsg(Message message, Long delayTimeMS, String key) {

      if (message == null)

         return;

         return null;

      if (key != null)

         message.setKey(key);

      if (delayTimeMS != null)

         message.setStartDeliverTime(System.currentTimeMillis() + delayTimeMS);// 10s后发送活跃消息

      producer.send(message);

      return producer.send(message);

   }



   /**

@@ -65,7 +66,8 @@
    * void 返回类型

    * @throws

    */

   public void sendTransactionalMsg(Message message, Long delayTimeMS, String key,ITransactionalMQEvent mqEvent) throws Exception{

   public void sendTransactionalMsg(Message message, Long delayTimeMS, String key, ITransactionalMQEvent mqEvent)

         throws Exception {

      if (key != null)

         message.setKey(key);

      if (delayTimeMS != null)

@@ -74,13 +76,13 @@
      orderTransactionProducer.send(message, new LocalTransactionExecuter() {

         @Override

         public TransactionStatus execute(Message arg0, Object arg1) {

            if(mqEvent!=null)

            if (mqEvent != null)

               mqEvent.excute();

            return TransactionStatus.CommitTransaction;

         }

      }, null);

   }

	


   /**

    * 发送事务消息

    * @Title: sendTransactionalMsg

@@ -92,12 +94,11 @@
    * void 返回类型

    * @throws

    */

   public void sendTransactionalMsg(Message message, String key,ITransactionalMQEvent mqEvent) throws Exception{

      sendTransactionalMsg(message,null,key,mqEvent);

   public void sendTransactionalMsg(Message message, String key, ITransactionalMQEvent mqEvent) throws Exception {

      sendTransactionalMsg(message, null, key, mqEvent);

   }

	
	
   interface ITransactionalMQEvent{



   interface ITransactionalMQEvent {

      public void excute();

   }

}