wangxianzhang
3 years ago
12 changed files with 341 additions and 30 deletions
@ -0,0 +1,60 @@ |
|||||
|
package com.epmet.commons.rocketmq.messages; |
||||
|
|
||||
|
import com.epmet.commons.tools.dto.form.mq.MqBaseFormDTO; |
||||
|
import lombok.Data; |
||||
|
|
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
|
||||
|
/** |
||||
|
* user库党员信息同步到partymember库的mq消息 |
||||
|
*/ |
||||
|
@Data |
||||
|
public class PartymemberSyncMQMsg extends MqBaseFormDTO { |
||||
|
|
||||
|
/** |
||||
|
* 党员列表，允许一次传输多个党员信息 |
||||
|
*/ |
||||
|
private List<PartyMemberSyncForm> partymemberList = new ArrayList<>(); |
||||
|
|
||||
|
@Data |
||||
|
public static class PartyMemberSyncForm { |
||||
|
|
||||
|
private String customerId; |
||||
|
private String agencyId; |
||||
|
private String agencyPids; |
||||
|
private String icResiUser; |
||||
|
private String name; |
||||
|
private String idCard; |
||||
|
private String mobile; |
||||
|
private String address; |
||||
|
private String rdsj; |
||||
|
private String sszb; |
||||
|
/** |
||||
|
* 是否流动党员 |
||||
|
*/ |
||||
|
private String isLd; |
||||
|
|
||||
|
/** |
||||
|
* 流动党员活动证号 |
||||
|
*/ |
||||
|
private String ldzh; |
||||
|
/** |
||||
|
* 职务 |
||||
|
*/ |
||||
|
private String partyZw; |
||||
|
/** |
||||
|
* 是否退休 |
||||
|
*/ |
||||
|
private String isTx; |
||||
|
/** |
||||
|
* 是否党员中心户 |
||||
|
*/ |
||||
|
private String isDyzxh; |
||||
|
/** |
||||
|
* 志愿者类型,逗号隔开 |
||||
|
*/ |
||||
|
private String volunteerCategory; |
||||
|
|
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,19 @@ |
|||||
|
package com.epmet.mq; |
||||
|
|
||||
|
import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants; |
||||
|
import com.epmet.commons.rocketmq.constants.TopicConstants; |
||||
|
import com.epmet.commons.rocketmq.register.MQAbstractRegister; |
||||
|
import com.epmet.commons.rocketmq.register.MQConsumerProperties; |
||||
|
import com.epmet.mq.listener.ResiPartyMemberSyncListener; |
||||
|
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
|
||||
|
@Component |
||||
|
public class RocketMQConsumerRegister extends MQAbstractRegister { |
||||
|
|
||||
|
@Override |
||||
|
public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { |
||||
|
register(consumerProperties, ConsomerGroupConstants.CREATE_RESI_PARTYMEMBER_SYNC_GROUP, MessageModel.CLUSTERING, |
||||
|
TopicConstants.PARTYMEMBER_RESI, "*", new ResiPartyMemberSyncListener()); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,106 @@ |
|||||
|
package com.epmet.mq.listener; |
||||
|
|
||||
|
import com.alibaba.fastjson.JSON; |
||||
|
import com.epmet.commons.rocketmq.constants.MQUserPropertys; |
||||
|
import com.epmet.commons.rocketmq.messages.PartymemberSyncMQMsg; |
||||
|
import com.epmet.commons.tools.exception.ExceptionUtils; |
||||
|
import com.epmet.commons.tools.exception.RenException; |
||||
|
import com.epmet.commons.tools.redis.RedisKeys; |
||||
|
import com.epmet.commons.tools.redis.RedisUtils; |
||||
|
import com.epmet.commons.tools.utils.ConvertUtils; |
||||
|
import com.epmet.commons.tools.utils.SpringContextUtils; |
||||
|
import com.epmet.modules.partymember.service.IcPartyMemberService; |
||||
|
import com.epmet.resi.partymember.dto.partymember.IcPartyMemberDTO; |
||||
|
import org.apache.commons.collections4.CollectionUtils; |
||||
|
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.slf4j.Logger; |
||||
|
import org.slf4j.LoggerFactory; |
||||
|
|
||||
|
import java.util.List; |
||||
|
|
||||
|
/** |
||||
|
* @author wxz |
||||
|
* @Description 创建党员居民信息消费者组，将user库的党员信息同步到partymember库的党员表。 |
||||
|
|
||||
|
* @return |
||||
|
* @date 2021.06.07 16:12 |
||||
|
*/ |
||||
|
public class ResiPartyMemberSyncListener implements MessageListenerConcurrently { |
||||
|
|
||||
|
private Logger logger = LoggerFactory.getLogger(getClass()); |
||||
|
|
||||
|
private RedisUtils redisUtils; |
||||
|
|
||||
|
@Override |
||||
|
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { |
||||
|
|
||||
|
if (redisUtils == null) { |
||||
|
redisUtils = SpringContextUtils.getBean(RedisUtils.class); |
||||
|
} |
||||
|
|
||||
|
try { |
||||
|
msgs.forEach(msg -> consumeMessage(msg)); |
||||
|
} catch (Exception e) { |
||||
|
logger.error(ExceptionUtils.getErrorStackTrace(e)); |
||||
|
return ConsumeConcurrentlyStatus.RECONSUME_LATER; |
||||
|
} |
||||
|
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* 逐条消费 |
||||
|
* @param messageExt |
||||
|
*/ |
||||
|
private void consumeMessage(MessageExt messageExt) { |
||||
|
//String tags = messageExt.getTags();
|
||||
|
String msg = new String(messageExt.getBody()); |
||||
|
String pendingMsgLabel = messageExt.getUserProperty(MQUserPropertys.BLOCKED_MSG_LABEL); |
||||
|
logger.info("居民-党员信息同步监听器-收到消息内容:{}", msg); |
||||
|
PartymemberSyncMQMsg msgObj = JSON.parseObject(msg, PartymemberSyncMQMsg.class); |
||||
|
|
||||
|
List<PartymemberSyncMQMsg.PartyMemberSyncForm> partymemberList = msgObj.getPartymemberList(); |
||||
|
if (CollectionUtils.isEmpty(partymemberList)) { |
||||
|
return; |
||||
|
} |
||||
|
partymemberList.stream().forEach(p -> { |
||||
|
IcPartyMemberDTO icPartyMemberDTO = ConvertUtils.sourceToTarget(p, IcPartyMemberDTO.class); |
||||
|
try { |
||||
|
SpringContextUtils.getBean(IcPartyMemberService.class).icPartyMemberSync(icPartyMemberDTO); |
||||
|
} catch (RenException e) { |
||||
|
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
||||
|
logger.error("【居民-党员信息同步】同步失败，居民id:{},错误信息:{}", p.getIcResiUser(), ExceptionUtils.getErrorStackTrace(e)); |
||||
|
} catch (Exception e) { |
||||
|
// 不是我们自己抛出的异常，可以让MQ重试
|
||||
|
logger.error("【居民-党员信息同步】同步失败居民id:{},错误信息:{}", p.getIcResiUser(), ExceptionUtils.getErrorStackTrace(e)); |
||||
|
throw e; |
||||
|
} |
||||
|
}); |
||||
|
|
||||
|
if (StringUtils.isNotBlank(pendingMsgLabel)) { |
||||
|
try { |
||||
|
removePendingMqMsgCache(pendingMsgLabel); |
||||
|
} catch (Exception e) { |
||||
|
logger.error("【居民-党员信息同步】-删除mq阻塞消息缓存失败:{}", ExceptionUtils.getErrorStackTrace(e)); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
logger.info("【居民-党员信息同步】处理完成，条数:{}", partymemberList.size()); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* @description |
||||
|
* |
||||
|
* @param pendingMsgLabel |
||||
|
* @return |
||||
|
* @author wxz |
||||
|
* @date 2021.10.14 16:32:32 |
||||
|
*/ |
||||
|
private void removePendingMqMsgCache(String pendingMsgLabel) { |
||||
|
String key = RedisKeys.blockedMqMsgKey(pendingMsgLabel); |
||||
|
Boolean rst = redisUtils.delete(key); |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue