rcoketmq集中管理

admin

2020-07-01 c54fb6a88876be994906d57d2d18e844686964d0

 fanli/src/main/java/com/yeshi/fanli/job/MQJob.java

@@ -9,11 +9,11 @@
import org.springframework.stereotype.Component;



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

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

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

import com.yeshi.fanli.entity.mq.MQUnSendInfo;

import com.yeshi.fanli.log.LogHelper;

import com.yeshi.fanli.service.inter.mq.MQUnSendInfoService;

import com.yeshi.fanli.service.manger.msg.RocketMQManager;

import com.yeshi.fanli.util.Constant;

import com.yeshi.fanli.util.StringUtil;



@@ -28,8 +28,10 @@


   @Resource

   private MQUnSendInfoService mqUnSendInfoService;

   @Resource(name = "producer")

   private Producer producer;

	
   @Resource

   private RocketMQManager rocketMQManager;





   public MQJob() {

   }

@@ -50,7 +52,7 @@
               msg.setKey(sendInfo.getKey());

            if (sendInfo.getDeliverTime() != null)

               msg.setStartDeliverTime(sendInfo.getDeliverTime().getTime());

            SendResult sendResult = producer.send(msg);

            SendResult sendResult =rocketMQManager.sendNormalMsg(msg, null,null);

            if (sendResult != null) {

               mqUnSendInfoService.deleteByPrimaryKey(sendInfo.getId());

               LogHelper.mqInfo("消息重发成功",sendResult.getMessageId(), sendInfo.getTopic(), sendInfo.getTag(), sendInfo.getBody());