java/mx_charging_pile.git

			@@ -5,6 +5,8 @@
			import com.ruoyi.integration.mongodb.service.PlatformStopChargingReplyService;
			import com.ruoyi.integration.rocket.model.PlatformStopChargingReplyMessage;
			import com.ruoyi.integration.rocket.util.EnhanceMessageHandler;
			import com.ruoyi.order.api.feignClient.ChargingOrderClient;
			import com.ruoyi.order.api.vo.PlatformStopChargingReplyVO;
			import lombok.extern.slf4j.Slf4j;
			import org.apache.rocketmq.spring.annotation.MessageModel;
			import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
			@@ -13,68 +15,37 @@
			import org.springframework.beans.factory.annotation.Autowired;
			import org.springframework.stereotype.Component;

			import javax.annotation.Resource;

			@Slf4j
			@Component
			@RocketMQMessageListener(
			messageModel = MessageModel.CLUSTERING,
			consumerGroup = "charge_platform_stop_charging_reply",
			topic = "charge_platform_stop_charging_reply",
			selectorExpression = "platform_stop_charging_reply",
			consumeThreadMax = 5 //默认是64个线程并发消息，配置 consumeThreadMax 参数指定并发消费线程数，避免太大导致资源不够
			selectorExpression = "platform_stop_charging_reply"
			)
			public class PlatformStopChargingReplyMessageListener extends EnhanceMessageHandler<PlatformStopChargingReplyMessage> implements RocketMQListener<PlatformStopChargingReplyMessage> {
			public class PlatformStopChargingReplyMessageListener implements RocketMQListener<PlatformStopChargingReplyMessage> {
			@Autowired
			private PlatformStopChargingReplyService platformStopChargingReplyService;

			@Override
			protected void handleMessage(PlatformStopChargingReplyMessage message) throws Exception {
			// 此时这里才是最终的业务处理，代码只需要处理资源类关闭异常，其他的可以交给父类重试
			log.info("远程停机命令回复-业务消息处理:{}",message);
			// 持久化消息
			PlatformStopChargingReply platformStopChargingReply = new PlatformStopChargingReply();
			BeanUtils.copyProperties(message,platformStopChargingReply);
			platformStopChargingReplyService.create(platformStopChargingReply);
			// 业务处理
			}

			@Override
			protected void handleMaxRetriesExceeded(PlatformStopChargingReplyMessage message) {
			// 当超过指定重试次数消息时此处方法会被调用
			// 生产中可以进行回退或其他业务操作
			log.error("消息消费失败，请执行后续处理");
			}


			/**
			* 是否执行重试机制
			*/
			@Override
			protected boolean isRetry() {
			return true;
			}

			@Override
			protected boolean throwException() {
			// 是否抛出异常，false搭配retry自行处理异常
			return false;
			}

			/**
			* 若需要处理消息过滤，在父级中进行统一处理，或者在此处实现之后，自行处理
			* @param message 待处理消息
			* @return true: 本次消息被过滤，false：不过滤
			*/
			@Override
			protected boolean filter(PlatformStopChargingReplyMessage message) {
			// 此处可做消息过滤
			return false;
			}
			@Resource
			private ChargingOrderClient chargingOrderClient;



			/**
			* 监听消费消息，不需要执行业务处理，委派给父类做基础操作，父类做完基础操作后会调用子类的实际处理类型
			*/
			@Override
			public void onMessage(PlatformStopChargingReplyMessage message) {
			super.dispatchMessage(message);
			log.info("远程停机命令回复-业务消息处理:{}",message);
			// 持久化消息
			PlatformStopChargingReply platformStopChargingReply = new PlatformStopChargingReply();
			BeanUtils.copyProperties(message,platformStopChargingReply);
			platformStopChargingReplyService.create(platformStopChargingReply);

			PlatformStopChargingReplyVO platformStopChargingReply1 = new PlatformStopChargingReplyVO();
			BeanUtils.copyProperties(platformStopChargingReply, platformStopChargingReply1);
			chargingOrderClient.terminateSuccessfulResponse(platformStopChargingReply1);
			}
			}