Browse Source
# Conflicts: # epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/constants/TopicConstants.java # epmet-module/epmet-message/epmet-message-client/pom.xml # epmet-module/epmet-message/epmet-message-client/src/main/java/com/epmet/constant/SystemMessageType.java # epmet-module/epmet-message/epmet-message-client/src/main/java/com/epmet/send/SendMqMsgUtil.javamaster
jianjun
4 years ago
16 changed files with 386 additions and 17 deletions
@ -0,0 +1,17 @@ |
|||
package com.epmet.commons.rocketmq.messages; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
|
|||
import java.io.Serializable; |
|||
|
|||
/** |
|||
* desc：项目变动通知消息实体类 |
|||
* @author liujianjun |
|||
*/ |
|||
@Data |
|||
@AllArgsConstructor |
|||
public class ProjectChangedMQMsg implements Serializable { |
|||
|
|||
private String customerId; |
|||
} |
|||
@ -0,0 +1,114 @@ |
|||
package com.epmet.mq; |
|||
|
|||
import com.alibaba.fastjson.JSON; |
|||
import com.epmet.commons.rocketmq.messages.ProjectChangedMQMsg; |
|||
import com.epmet.commons.tools.distributedlock.DistributedLock; |
|||
import com.epmet.commons.tools.exception.RenException; |
|||
import com.epmet.commons.tools.utils.SpringContextUtils; |
|||
import com.epmet.dto.extract.form.ExtractOriginFormDTO; |
|||
import com.epmet.service.evaluationindex.extract.todata.FactOriginExtractService; |
|||
import com.epmet.service.evaluationindex.extract.toscreen.ScreenExtractService; |
|||
import com.epmet.util.DimIdGenerator; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; |
|||
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; |
|||
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; |
|||
import org.apache.rocketmq.common.message.MessageExt; |
|||
import org.redisson.api.RLock; |
|||
import org.slf4j.Logger; |
|||
import org.slf4j.LoggerFactory; |
|||
|
|||
import javax.annotation.PreDestroy; |
|||
import java.util.Date; |
|||
import java.util.List; |
|||
import java.util.concurrent.ExecutionException; |
|||
import java.util.concurrent.Future; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.stream.Collectors; |
|||
|
|||
/** |
|||
* @Description 项目变动-监听器 |
|||
* @return |
|||
* @author wxz |
|||
* @date 2021.03.03 16:10 |
|||
*/ |
|||
@Slf4j |
|||
public class ProjectChangedCustomListener implements MessageListenerConcurrently { |
|||
|
|||
private Logger logger = LoggerFactory.getLogger(getClass()); |
|||
|
|||
@Override |
|||
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { |
|||
long start = System.currentTimeMillis(); |
|||
try { |
|||
List<String> customerIds = msgs.stream().map(messageExt -> new String(messageExt.getBody())).distinct().collect(Collectors.toList()); |
|||
customerIds.forEach(this::consumeMessage); |
|||
} catch (Exception e) { |
|||
//失败不重发
|
|||
logger.error("consumeMessage fail,msg:{}",e.getMessage()); |
|||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
|||
} |
|||
log.info("consumeMessage success, cost:{} ms",System.currentTimeMillis() - start); |
|||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
|||
} |
|||
|
|||
private void consumeMessage(String msg) { |
|||
logger.info("receive customerId:{}", JSON.toJSONString(msg)); |
|||
ProjectChangedMQMsg msgObj = JSON.parseObject(msg, ProjectChangedMQMsg.class); |
|||
|
|||
DistributedLock distributedLock = null; |
|||
RLock lock = null; |
|||
try { |
|||
String customerId = msgObj.getCustomerId(); |
|||
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
|||
lock = distributedLock.getLock(String.format("lock:project_changed:%s", customerId) |
|||
,30L, 30L, TimeUnit.SECONDS); |
|||
|
|||
if (StringUtils.isBlank(customerId)){ |
|||
logger.error("consumer project_changed fail,msg:{}",customerId); |
|||
return; |
|||
} |
|||
//消息被消费太快 业务数据还没有完成 歇一会先
|
|||
try { |
|||
Thread.sleep(60L); |
|||
} catch (InterruptedException e) { |
|||
logger.error("consumeMessage sleep exception",e); |
|||
} |
|||
ExtractOriginFormDTO extractOriginFormDTO = new ExtractOriginFormDTO(); |
|||
extractOriginFormDTO.setCustomerId(customerId); |
|||
|
|||
String dateId = DimIdGenerator.getDateDimId(new Date()); |
|||
extractOriginFormDTO.setDateId(dateId); |
|||
Future<?> aBoolean = SpringContextUtils.getBean(FactOriginExtractService.class).submitProjectRelationData(extractOriginFormDTO,null); |
|||
try { |
|||
aBoolean.get(); |
|||
} catch (InterruptedException e) { |
|||
e.printStackTrace(); |
|||
} catch (ExecutionException e) { |
|||
e.printStackTrace(); |
|||
} |
|||
if (aBoolean.isDone()){ |
|||
SpringContextUtils.getBean(ScreenExtractService.class).extractPartData(customerId,dateId); |
|||
} |
|||
logger.info("consumer projectChanged msg success,{}",aBoolean); |
|||
} catch (RenException e) { |
|||
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
|||
logger.error("【RocketMQ】消费项目变动消息失败:",e); |
|||
} catch (Exception e) { |
|||
// 不是我们自己抛出的异常，可以让MQ重试
|
|||
logger.error("【RocketMQ】消费项目变动消息异常:",e); |
|||
throw e; |
|||
} finally { |
|||
if (distributedLock != null){ |
|||
distributedLock.unLock(lock); |
|||
} |
|||
} |
|||
} |
|||
@PreDestroy |
|||
public void saveCalStatus() { |
|||
//todo
|
|||
log.info("data-statical-server服务被关闭，执行未执行完的动作"); |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,68 @@ |
|||
package com.epmet.mq; |
|||
|
|||
import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants; |
|||
import com.epmet.commons.rocketmq.constants.TopicConstants; |
|||
import com.epmet.commons.tools.constant.NumConstant; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; |
|||
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; |
|||
import org.apache.rocketmq.client.exception.MQClientException; |
|||
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.core.annotation.Order; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
@Order(value = 111) |
|||
public class RocketMQConsumerRegister { |
|||
|
|||
@Value("${rocketmq.name-server}") |
|||
private String nameServer; |
|||
|
|||
/** |
|||
* @return |
|||
* @Description 注册监听器 |
|||
* @author wxz |
|||
* @date 2021.03.03 16:09 |
|||
*/ |
|||
@PostConstruct |
|||
public void registerAllListeners() { |
|||
try { |
|||
register(ConsomerGroupConstants.PROJECT_CHANGED_COMPONENTS_GROUP, MessageModel.CLUSTERING, TopicConstants.PROJECT_CHANGED, "*", new ProjectChangedCustomListener()); |
|||
} catch (MQClientException e) { |
|||
log.error("registerAllListeners exception", e); |
|||
} |
|||
} |
|||
|
|||
public void register(String group, MessageModel messageModel, String topic, String subException, MessageListenerConcurrently listener) throws MQClientException { |
|||
// 实例化消费者
|
|||
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group); |
|||
|
|||
// 设置NameServer的地址
|
|||
consumer.setNamesrvAddr(nameServer); |
|||
consumer.setMessageModel(messageModel); |
|||
consumer.setInstanceName(buildInstanceName()); |
|||
// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
|
|||
consumer.subscribe(topic, subException); |
|||
// 注册回调实现类来处理从broker拉取回来的消息
|
|||
consumer.registerMessageListener(listener); |
|||
//一次批量拉去10条消息
|
|||
consumer.setConsumeMessageBatchMaxSize(NumConstant.TEN); |
|||
// 启动消费者实例
|
|||
consumer.start(); |
|||
} |
|||
|
|||
private String buildInstanceName() { |
|||
String instanceName = ""; |
|||
for (int i = 0; i < 4; i++) { |
|||
int t = (int) (Math.random() * 10); |
|||
instanceName = instanceName.concat(t + ""); |
|||
} |
|||
|
|||
return instanceName; |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue