forked from luyan/epmet-cloud-lingshan
wxz
5 years ago
7 changed files with 346 additions and 25 deletions
@ -0,0 +1,62 @@ |
|||
package com.epmet.mq; |
|||
|
|||
import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants; |
|||
import com.epmet.commons.rocketmq.constants.TopicConstants; |
|||
import com.epmet.mq.listener.InitCustomerOrgRolesListener; |
|||
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; |
|||
import org.apache.rocketmq.client.consumer.listener.MessageListener; |
|||
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.stereotype.Component; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
|
|||
@Component |
|||
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(nameServer, ConsomerGroupConstants.INIT_CUSTOMER_ORG_ROLES_GROUP, MessageModel.CLUSTERING, TopicConstants.INIT_CUSTOMER, "*", new InitCustomerOrgRolesListener()); |
|||
} catch (MQClientException e) { |
|||
e.printStackTrace(); |
|||
} |
|||
} |
|||
|
|||
public void register(String nameServer, String group, MessageModel messageModel, String topic, String subException, MessageListener 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); |
|||
// 启动消费者实例
|
|||
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; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,102 @@ |
|||
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.commons.tools.utils.SpringContextUtils; |
|||
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.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 java.util.List; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* @Description 创建客户-组织和角色监听器 |
|||
* @return |
|||
* @author wxz |
|||
* @date 2021.03.03 16:10 |
|||
*/ |
|||
public class InitCustomerOrgRolesListener implements MessageListenerConcurrently { |
|||
|
|||
private Logger logger = LoggerFactory.getLogger(getClass()); |
|||
|
|||
@Override |
|||
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { |
|||
try { |
|||
msgs.forEach(msg -> consumeMessage(msg)); |
|||
} catch (Exception e) { |
|||
logger.error(ExceptionUtils.getErrorStackTrace(e)); |
|||
return ConsumeConcurrentlyStatus.RECONSUME_LATER; |
|||
} |
|||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
|||
} |
|||
|
|||
private void consumeMessage(MessageExt messageExt) { |
|||
String msg = new String(messageExt.getBody()); |
|||
logger.info("初始化客户-初始化组织信息-收到消息内容:{}", msg); |
|||
InitCustomerMQMsg msgObj = JSON.parseObject(msg, InitCustomerMQMsg.class); |
|||
|
|||
DistributedLock distributedLock = null; |
|||
RLock lock = null; |
|||
try { |
|||
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
|||
lock = distributedLock.getLock(String.format("lock:init_customer_org:%s", msgObj.getCustomerId()), |
|||
30l, 30l, TimeUnit.SECONDS); |
|||
SpringContextUtils.getBean(AgencyService.class).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; |
|||
} |
|||
} |
|||
@ -0,0 +1,62 @@ |
|||
package com.epmet.mq; |
|||
|
|||
import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants; |
|||
import com.epmet.commons.rocketmq.constants.TopicConstants; |
|||
import com.epmet.mq.listener.InitCustomerComponentsListener; |
|||
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; |
|||
import org.apache.rocketmq.client.consumer.listener.MessageListener; |
|||
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.stereotype.Component; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
|
|||
@Component |
|||
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(nameServer, ConsomerGroupConstants.INIT_CUSTOMER_COMPONENTS_GROUP, MessageModel.CLUSTERING, TopicConstants.INIT_CUSTOMER, "*", new InitCustomerComponentsListener()); |
|||
} catch (MQClientException e) { |
|||
e.printStackTrace(); |
|||
} |
|||
} |
|||
|
|||
public void register(String nameServer, String group, MessageModel messageModel, String topic, String subException, MessageListener 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); |
|||
// 启动消费者实例
|
|||
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; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,68 @@ |
|||
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.commons.tools.utils.SpringContextUtils; |
|||
import com.epmet.dto.CustomerHomeDTO; |
|||
import com.epmet.service.CustomerHomeService; |
|||
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 java.util.List; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* 监听初始化客户动作，为客户初始化角色列表 |
|||
*/ |
|||
public class InitCustomerComponentsListener implements MessageListenerConcurrently { |
|||
|
|||
private Logger logger = LoggerFactory.getLogger(getClass()); |
|||
|
|||
@Override |
|||
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { |
|||
try { |
|||
msgs.forEach(msg -> consumeMessage(msg)); |
|||
} catch (Exception e) { |
|||
logger.error(ExceptionUtils.getErrorStackTrace(e)); |
|||
return ConsumeConcurrentlyStatus.RECONSUME_LATER; |
|||
} |
|||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
|||
} |
|||
|
|||
private void consumeMessage(MessageExt messageExt) { |
|||
String msg = new String(messageExt.getBody()); |
|||
logger.info("初始化客户-初始化客户自定义信息-收到消息内容:{}", msg); |
|||
InitCustomerMQMsg msgObj = JSON.parseObject(msg, InitCustomerMQMsg.class); |
|||
|
|||
CustomerHomeDTO customerHomeDTO = new CustomerHomeDTO(); |
|||
customerHomeDTO.setCustomerId(msgObj.getCustomerId()); |
|||
|
|||
DistributedLock distributedLock = null; |
|||
RLock lock = null; |
|||
try { |
|||
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
|||
lock = distributedLock.getLock(String.format("lock:init_customer_home:%s", msgObj.getCustomerId()), |
|||
30l, 30l, TimeUnit.SECONDS); |
|||
|
|||
CustomerHomeService customerHomeService = SpringContextUtils.getBean(CustomerHomeService.class); |
|||
customerHomeService.init(customerHomeDTO); |
|||
} 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); |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue