jianjun
4 years ago
10 changed files with 186 additions and 148 deletions
@ -0,0 +1,43 @@ |
|||||
|
package com.epmet.commons.rocketmq.register; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; |
||||
|
|
||||
|
import java.io.Serializable; |
||||
|
|
||||
|
/** |
||||
|
* desc:mq 消费配置类 |
||||
|
* |
||||
|
* @author: LiuJanJun |
||||
|
* @date: 2021/4/30 2:39 下午 |
||||
|
* @version: 1.0 |
||||
|
*/ |
||||
|
@Data |
||||
|
public class ConsumerConfigProperties implements Serializable { |
||||
|
|
||||
|
private static final long serialVersionUID = 2069676324708473773L; |
||||
|
/** |
||||
|
* 消费者组 |
||||
|
*/ |
||||
|
private String consumerGroup; |
||||
|
/** |
||||
|
* 主题 |
||||
|
*/ |
||||
|
private String topic; |
||||
|
/** |
||||
|
* 标签 |
||||
|
*/ |
||||
|
private String tag = "*"; |
||||
|
/** |
||||
|
* 最小消费的线程数 |
||||
|
*/ |
||||
|
private int consumeThreadMin = 2; |
||||
|
/** |
||||
|
* 最大消费的线程数 |
||||
|
*/ |
||||
|
private int consumeThreadMax = 4; |
||||
|
/** |
||||
|
* 消费监听器 |
||||
|
*/ |
||||
|
private MessageListenerConcurrently consumerListener; |
||||
|
} |
||||
@ -0,0 +1,79 @@ |
|||||
|
package com.epmet.commons.rocketmq.register; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; |
||||
|
import org.apache.rocketmq.client.exception.MQClientException; |
||||
|
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
|
||||
|
import javax.annotation.PostConstruct; |
||||
|
|
||||
|
/** |
||||
|
* desc：注册mq监听器 |
||||
|
* |
||||
|
* @author liujianjun |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
public abstract class MQConsumerRegister { |
||||
|
@Value("${spring.profiles.active}") |
||||
|
private String env; |
||||
|
@Value("${rocketmq.name-server}") |
||||
|
private String namesrvAddr; |
||||
|
|
||||
|
public abstract ConsumerConfigProperties getConsumerProperty(); |
||||
|
|
||||
|
|
||||
|
/** |
||||
|
* @return |
||||
|
* @Description 注册监听器 |
||||
|
* @author wxz |
||||
|
* @date 2021.03.03 16:09 |
||||
|
*/ |
||||
|
@PostConstruct |
||||
|
public void registerMQListener() { |
||||
|
ConsumerConfigProperties consumerProperty = getConsumerProperty(); |
||||
|
log.info("registerAllListeners consumers:{} success", consumerProperty); |
||||
|
//本地环境不注册
|
||||
|
if ("local".equals(env)) { |
||||
|
try { |
||||
|
// 实例化消费者
|
||||
|
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerProperty.getConsumerGroup()); |
||||
|
|
||||
|
// 设置NameServer的地址
|
||||
|
consumer.setNamesrvAddr(namesrvAddr); |
||||
|
consumer.setMessageModel(MessageModel.CLUSTERING); |
||||
|
consumer.setInstanceName(buildInstanceName()); |
||||
|
// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
|
||||
|
consumer.subscribe(consumer.getConsumerGroup(), consumerProperty.getTag()); |
||||
|
// 注册回调实现类来处理从broker拉取回来的消息
|
||||
|
consumer.registerMessageListener(consumerProperty.getConsumerListener()); |
||||
|
consumer.setConsumeThreadMin(consumerProperty.getConsumeThreadMin()); |
||||
|
consumer.setConsumeThreadMax(consumerProperty.getConsumeThreadMax()); |
||||
|
// 启动消费者实例
|
||||
|
consumer.start(); |
||||
|
} catch (MQClientException e) { |
||||
|
log.info("registerMQListener exception", e); |
||||
|
} |
||||
|
|
||||
|
} |
||||
|
|
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* desc: 因为docker-compose部署有问题 所有自己命名 |
||||
|
* |
||||
|
* @param |
||||
|
* @return java.lang.String |
||||
|
* @author LiuJanJun |
||||
|
* @date 2021/4/30 5:00 下午 |
||||
|
*/ |
||||
|
private String buildInstanceName() { |
||||
|
String instanceName = ""; |
||||
|
for (int i = 0; i < 4; i++) { |
||||
|
int t = (int) (Math.random() * 10); |
||||
|
instanceName = instanceName.concat(t + ""); |
||||
|
} |
||||
|
|
||||
|
return instanceName; |
||||
|
} |
||||
|
} |
||||
@ -1,95 +0,0 @@ |
|||||
package com.epmet.mq.listener; |
|
||||
|
|
||||
import com.alibaba.fastjson.JSON; |
|
||||
import com.epmet.commons.rocketmq.messages.InitCustomerMQMsg; |
|
||||
import com.epmet.commons.tools.distributedlock.DistributedLock; |
|
||||
import com.epmet.commons.tools.exception.ExceptionUtils; |
|
||||
import com.epmet.commons.tools.exception.RenException; |
|
||||
import com.epmet.constant.UserWorkType; |
|
||||
import com.epmet.dto.CustomerAgencyDTO; |
|
||||
import com.epmet.dto.form.AddAgencyAndStaffFormDTO; |
|
||||
import com.epmet.dto.form.AdminStaffFromDTO; |
|
||||
import com.epmet.service.AgencyService; |
|
||||
import org.apache.rocketmq.common.message.MessageExt; |
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener; |
|
||||
import org.redisson.api.RLock; |
|
||||
import org.slf4j.Logger; |
|
||||
import org.slf4j.LoggerFactory; |
|
||||
import org.springframework.beans.factory.annotation.Autowired; |
|
||||
|
|
||||
import java.util.concurrent.TimeUnit; |
|
||||
|
|
||||
/** |
|
||||
* 监听初始化客户动作，为客户初始化角色列表 |
|
||||
*/ |
|
||||
//@RocketMQMessageListener(topic = TopicConstants.INIT_CUSTOMER,
|
|
||||
// consumerGroup = ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP,
|
|
||||
// messageModel = MessageModel.CLUSTERING,
|
|
||||
// selectorExpression = "*")
|
|
||||
//@Component
|
|
||||
public class InitCustomerOrgListener implements RocketMQListener<MessageExt> { |
|
||||
|
|
||||
private Logger logger = LoggerFactory.getLogger(getClass()); |
|
||||
|
|
||||
@Autowired |
|
||||
private AgencyService agencyService; |
|
||||
|
|
||||
@Autowired |
|
||||
private DistributedLock distributedLock; |
|
||||
|
|
||||
@Override |
|
||||
public void onMessage(MessageExt messageExt) { |
|
||||
String msg = new String(messageExt.getBody()); |
|
||||
logger.info("初始化客户-初始化组织信息-收到消息内容:{}", msg); |
|
||||
InitCustomerMQMsg msgObj = JSON.parseObject(msg, InitCustomerMQMsg.class); |
|
||||
|
|
||||
RLock lock = null; |
|
||||
try { |
|
||||
lock = distributedLock.getLock(String.format("lock:init_customer_org:%s", msgObj.getCustomerId()), |
|
||||
30L, 30l, TimeUnit.SECONDS); |
|
||||
agencyService.saveRootAgency(constructRootAndAgencyDTO(msgObj)); |
|
||||
} catch (RenException e) { |
|
||||
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
|
||||
logger.error("【RocketMQ】初始化客户组织失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
|
||||
} catch (Exception e) { |
|
||||
// 不是我们自己抛出的异常，可以让MQ重试
|
|
||||
logger.error("【RocketMQ】初始化客户组织失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
|
||||
throw e; |
|
||||
} finally { |
|
||||
distributedLock.unLock(lock); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
/** |
|
||||
* @Description 构造dto |
|
||||
* @return |
|
||||
* @author wxz |
|
||||
* @date 2021.01.06 15:03 |
|
||||
*/ |
|
||||
private AddAgencyAndStaffFormDTO constructRootAndAgencyDTO(InitCustomerMQMsg msgObj) { |
|
||||
AddAgencyAndStaffFormDTO agencyAndStaff = new AddAgencyAndStaffFormDTO(); |
|
||||
//客户组织信息
|
|
||||
CustomerAgencyDTO agencyDTO = new CustomerAgencyDTO(); |
|
||||
agencyDTO.setId(msgObj.getAgency().getAgencyId()); |
|
||||
agencyDTO.setCustomerId(msgObj.getCustomerId()); |
|
||||
agencyDTO.setOrganizationName(msgObj.getAgency().getOrganizationName()); |
|
||||
agencyDTO.setLevel(msgObj.getAgency().getLevel()); |
|
||||
agencyDTO.setAreaCode(msgObj.getAgency().getAreaCode()); |
|
||||
agencyDTO.setProvince(msgObj.getAgency().getProvince()); |
|
||||
agencyDTO.setCity(msgObj.getAgency().getCity()); |
|
||||
agencyDTO.setDistrict(msgObj.getAgency().getDistrict()); |
|
||||
agencyAndStaff.setAgencyDTO(agencyDTO); |
|
||||
|
|
||||
//客户管理员信息
|
|
||||
AdminStaffFromDTO staffSubmitFrom = new AdminStaffFromDTO(); |
|
||||
staffSubmitFrom.setCustomerId(msgObj.getCustomerId()); |
|
||||
staffSubmitFrom.setAgencyId(msgObj.getStaff().getAgencyId()); |
|
||||
staffSubmitFrom.setGender(msgObj.getStaff().getGender()); |
|
||||
staffSubmitFrom.setMobile(msgObj.getStaff().getMobile()); |
|
||||
staffSubmitFrom.setName(msgObj.getStaff().getName()); |
|
||||
staffSubmitFrom.setWorkType(UserWorkType.FULL_TIME); |
|
||||
agencyAndStaff.setStaffDTO(staffSubmitFrom); |
|
||||
|
|
||||
return agencyAndStaff; |
|
||||
} |
|
||||
} |
|
||||
@ -1,36 +0,0 @@ |
|||||
package com.epmet.mq; |
|
||||
|
|
||||
import com.alibaba.fastjson.JSON; |
|
||||
import com.epmet.commons.rocketmq.messages.GroupAchievementMQMsg; |
|
||||
import com.epmet.modules.group.service.StatsAchievementService; |
|
||||
import lombok.extern.slf4j.Slf4j; |
|
||||
import org.apache.commons.lang3.StringUtils; |
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener; |
|
||||
import org.springframework.beans.factory.annotation.Autowired; |
|
||||
|
|
||||
/** |
|
||||
* desc: 小组成就mq监听 |
|
||||
* |
|
||||
* @author: LiuJanJun |
|
||||
* @date: 2021/4/22 10:10 下午 |
|
||||
* @version: 1.0 |
|
||||
*/ |
|
||||
@Slf4j |
|
||||
//@Component
|
|
||||
//@RocketMQMessageListener(topic = "group_achievement", consumerGroup = ConsomerGroupConstants.GROUP_ACHIEVEMENT_COMPONENTS_GROUP)
|
|
||||
public class GroupAchieventConsumerListener implements RocketMQListener<GroupAchievementMQMsg> { |
|
||||
@Autowired |
|
||||
private StatsAchievementService statsAchievementService; |
|
||||
|
|
||||
@Override |
|
||||
public void onMessage(GroupAchievementMQMsg msg) { |
|
||||
log.info("receive msg:{}", JSON.toJSONString(msg)); |
|
||||
if (StringUtils.isBlank(msg.getAchievementType()) || StringUtils.isBlank(msg.getCustomerId()) |
|
||||
|| StringUtils.isBlank(msg.getGroupId())) { |
|
||||
log.error("consumer groupAchievement fail,msg:{}", msg); |
|
||||
return; |
|
||||
} |
|
||||
Boolean aBoolean = statsAchievementService.calculateAcm(msg.getCustomerId(), msg.getGroupId(), msg.getAchievementType()); |
|
||||
log.info("consumer groupAchievement msg success,{}", aBoolean); |
|
||||
} |
|
||||
} |
|
||||
Loading…
Reference in new issue