Skip to content

张云鹏 / uc-consumer

Go to a project
Toggle navigation
Toggle navigation pinning
Commit f4ad8dc9 f4ad8dc9623f655c9707d6d52e27081434f08de1 by xianghan
Options

1.同步master，优化任务处理过程

1 parent 60b8b824
Inline Side-by-side
Showing 10 changed files with 89 additions and 20 deletions
...@@ -5,6 +5,7 @@ import org.springframework.data.jpa.repository.JpaRepository; ...@@ -5,6 +5,7 @@ import org.springframework.data.jpa.repository.JpaRepository;
5 import org.springframework.data.jpa.repository.JpaSpecificationExecutor; 5 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
6 import org.springframework.data.jpa.repository.Modifying; 6 import org.springframework.data.jpa.repository.Modifying;
7 import org.springframework.data.jpa.repository.Query; 7 import org.springframework.data.jpa.repository.Query;
8 import org.springframework.data.repository.query.Param;
8 9
9 import java.time.LocalDateTime; 10 import java.time.LocalDateTime;
10 import java.util.List; 11 import java.util.List;
...@@ -23,6 +24,23 @@ public interface MemberRepository extends JpaRepository<Member, Long>, JpaSpecif ...@@ -23,6 +24,23 @@ public interface MemberRepository extends JpaRepository<Member, Long>, JpaSpecif
23 Optional<Member> findByIdOrCode(Long id, String code); 24 Optional<Member> findByIdOrCode(Long id, String code);
24 25
25 @Modifying 26 @Modifying
26 @Query(value = "UPDATE `uc_user_tv` SET `user_iptv_id` = ?2, `update_time` = ?3 WHERE `id` = ?1", nativeQuery = true) 27 @Query(value = "UPDATE `uc_member` SET `user_iptv_id` = ?2, `update_time` = ?3 , `bind_iptv_platform_type`= 0, " +
28 "`bind_iptv_time`=?3 WHERE `id` = ?1", nativeQuery = true)
27 void updateUserIptvIdById(Long id, Long userIptvId, LocalDateTime now); 29 void updateUserIptvIdById(Long id, Long userIptvId, LocalDateTime now);
30
31 @Modifying
32 @Query(value = "UPDATE `uc_member` SET `exp` = :#{#resources.exp}, `level` = :#{#resources.level} , `update_time`= now() " +
33 " WHERE `id` = :#{#resources.id}", nativeQuery = true)
34 void updateExpAndLevel(@Param("resources") Member member);
35
36 @Modifying
37 @Query(value = "UPDATE `uc_member` SET `points` = :#{#resources.points}, `due_points` = :#{#resources.duePoints} , `update_time`= now() " +
38 " WHERE `id` = :#{#resources.id}", nativeQuery = true)
39 void updatePointAndDuePoint(@Param("resources") Member resources);
40
41 @Modifying
42 @Query(value = "UPDATE `uc_member` SET `coupon_amount` = :#{#resources.couponAmount}, `due_coupon_amount` = :#{#resources.dueCouponAmount} , `update_time`= now() " +
43 " WHERE `id` = :#{#resources.id}", nativeQuery = true)
44 void doUpdateMemberCoupon(@Param("resources") Member member);
45
28 } 46 }
......
...@@ -86,4 +86,6 @@ public interface MemberService { ...@@ -86,4 +86,6 @@ public interface MemberService {
86 void unbind(Member resources); 86 void unbind(Member resources);
87 87
88 void updateUserIptvIdById(Long id, Long userIptvId, LocalDateTime now); 88 void updateUserIptvIdById(Long id, Long userIptvId, LocalDateTime now);
89
90 void doUpdateMemberCoupon(Member member);
89 } 91 }
......
...@@ -111,12 +111,22 @@ public class MemberServiceImpl implements MemberService { ...@@ -111,12 +111,22 @@ public class MemberServiceImpl implements MemberService {
111 } 111 }
112 112
113 @Override 113 @Override
114 @Transactional(rollbackFor = Exception.class)
114 public MemberDTO doUpdateMemberExpAndLevel(Member resources) { 115 public MemberDTO doUpdateMemberExpAndLevel(Member resources) {
115 MemberDTO memberDTO = this.update(resources); 116 MemberDTO memberDTO = this.findByCode(resources.getCode());
117
118 Member member = new Member();
119 member.setId(memberDTO.getId());
120 member.setExp(resources.getExp());
121 member.setLevel(resources.getLevel());
122 member.setUpdateTime(LocalDateTime.now());
123 this.memberRepository.updateExpAndLevel(member);
124 // MemberDTO memberDTO = this.update(resources);
116 return memberDTO; 125 return memberDTO;
117 } 126 }
118 127
119 @Override 128 @Override
129 @Transactional(rollbackFor = Exception.class)
120 public MemberDTO unbindUserIpTv(Member member) { 130 public MemberDTO unbindUserIpTv(Member member) {
121 Member _member = this.save(member); 131 Member _member = this.save(member);
122 return this.memberMapper.toDto(_member); 132 return this.memberMapper.toDto(_member);
...@@ -178,6 +188,12 @@ public class MemberServiceImpl implements MemberService { ...@@ -178,6 +188,12 @@ public class MemberServiceImpl implements MemberService {
178 188
179 @Override 189 @Override
180 @Transactional(rollbackFor = Exception.class) 190 @Transactional(rollbackFor = Exception.class)
191 public void doUpdateMemberCoupon(Member member) {
192 this.memberRepository.doUpdateMemberCoupon(member);
193 }
194
195 @Override
196 @Transactional(rollbackFor = Exception.class)
181 public MemberDTO update(Member resources) { 197 public MemberDTO update(Member resources) {
182 198
183 log.info("MemberServiceImpl ==>> update ==>> resources ==>> [{}]" , resources); 199 log.info("MemberServiceImpl ==>> update ==>> resources ==>> [{}]" , resources);
...@@ -219,11 +235,11 @@ public class MemberServiceImpl implements MemberService { ...@@ -219,11 +235,11 @@ public class MemberServiceImpl implements MemberService {
219 235
220 Member member = this.memberRepository.findById(resources.getId()).orElseGet(Member::new); 236 Member member = this.memberRepository.findById(resources.getId()).orElseGet(Member::new);
221 ValidationUtil.isNull(member.getId(), "Member", "id", resources.getId()); 237 ValidationUtil.isNull(member.getId(), "Member", "id", resources.getId());
222 member.copy(resources); 238 this.memberRepository.updatePointAndDuePoint(resources);
223 239 // member.copy(resources);
224 Member _member = this.save(member);
225 240
226 return this.memberMapper.toDto(_member); 241 // Member _member = this.save(member);
242 return this.memberMapper.toDto(member);
227 243
228 } catch (Exception e) { 244 } catch (Exception e) {
229 e.printStackTrace(); 245 e.printStackTrace();
......
...@@ -25,7 +25,7 @@ public class CouponOperationServiceImpl implements CouponOperationService { ...@@ -25,7 +25,7 @@ public class CouponOperationServiceImpl implements CouponOperationService {
25 String memberCode = member.getCode(); 25 String memberCode = member.getCode();
26 MemberDTO memberDTO = this.memberService.findByCode(memberCode); 26 MemberDTO memberDTO = this.memberService.findByCode(memberCode);
27 member.setId(memberDTO.getId()); 27 member.setId(memberDTO.getId());
28 this.memberOperationService.doUpdateMember(member); 28 this.memberService.doUpdateMemberCoupon(member);
29 } 29 }
30 30
31 public void asyncCouponHistory(CouponHistory couponHistory) { 31 public void asyncCouponHistory(CouponHistory couponHistory) {
......
...@@ -94,7 +94,15 @@ public class MemberOperationServiceImpl implements MemberOperationService { ...@@ -94,7 +94,15 @@ public class MemberOperationServiceImpl implements MemberOperationService {
94 // @CachePut(key = "#resources.id") 94 // @CachePut(key = "#resources.id")
95 @Override 95 @Override
96 public MemberDTO doUpdateMemberPoints(Member resources) { 96 public MemberDTO doUpdateMemberPoints(Member resources) {
97 return this.memberService.doUpdateMemberPoints(resources); 97 // return this.memberService.doUpdateMemberPoints(resources);
98 MemberDTO memberDTO = this.findByCode(resources.getCode());
99 Member member = new Member();
100 member.setId(memberDTO.getId());
101 member.setPoints(resources.getPoints());
102 member.setDuePoints(resources.getDuePoints());
103 member.setUpdateTime(LocalDateTime.now());
104 MemberDTO memberDTO1 = this.memberService.doUpdateMemberPoints(member);
105 return memberDTO1;
98 } 106 }
99 107
100 @Override 108 @Override
......
...@@ -72,14 +72,14 @@ public class UcEventBusIptv2ManagementUcEngine { ...@@ -72,14 +72,14 @@ public class UcEventBusIptv2ManagementUcEngine {
72 @RabbitListener(queues = "#{rabbitMqSourceConfig.getEventBusQueue()}", 72 @RabbitListener(queues = "#{rabbitMqSourceConfig.getEventBusQueue()}",
73 containerFactory = "#{rabbitMqSourceConfig.getEventBusSource()}", 73 containerFactory = "#{rabbitMqSourceConfig.getEventBusSource()}",
74 autoStartup = "#{rabbitMqSourceConfig.getEventBusStartUp()}", 74 autoStartup = "#{rabbitMqSourceConfig.getEventBusStartUp()}",
75 ackMode = "MANUAL") 75 ackMode = "AUTO")
76 public void eventBusConsumer(Channel channel, Message message, String content) throws ParseException, IOException { 76 public void eventBusConsumer(Channel channel, Message message, String content) throws ParseException, IOException {
77 log.info(" receive dataSync msg , content is : {} ", content); 77 log.info(" receive dataSync msg , content is : {} ", content);
78 try { 78 try {
79 79
80 this.parseContent(content); 80 this.parseContent(content);
81 81
82 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); 82 // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
83 83
84 } catch (Exception e) { 84 } catch (Exception e) {
85 85
...@@ -322,7 +322,7 @@ public class UcEventBusIptv2ManagementUcEngine { ...@@ -322,7 +322,7 @@ public class UcEventBusIptv2ManagementUcEngine {
322 description.put("mediaCode", mediaCode); 322 description.put("mediaCode", mediaCode);
323 description.put("time", time); 323 description.put("time", time);
324 msgData1.setDescription(JSON.toJSONString(description)); 324 msgData1.setDescription(JSON.toJSONString(description));
325 dataSyncMsg.setMsg(msgData1); 325 dataSyncMsg.setMsgData(JSONObject.toJSONString(msgData1));
326 return dataSyncMsg; 326 return dataSyncMsg;
327 } 327 }
328 328
......
...@@ -61,8 +61,9 @@ public class UcGatewayIptv2IptvConsumer { ...@@ -61,8 +61,9 @@ public class UcGatewayIptv2IptvConsumer {
61 log.info(" receive dataSync msg , content is : {} ", content); 61 log.info(" receive dataSync msg , content is : {} ", content);
62 try { 62 try {
63 DataSyncMsg dataSyncMsg = this.parseContent(content); 63 DataSyncMsg dataSyncMsg = this.parseContent(content);
64 64 if (Objects.nonNull(dataSyncMsg)) {
65 this.taskDeal(dataSyncMsg); 65 this.taskDeal(dataSyncMsg);
66 }
66 67
67 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); 68 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
68 69
...@@ -95,7 +96,13 @@ public class UcGatewayIptv2IptvConsumer { ...@@ -95,7 +96,13 @@ public class UcGatewayIptv2IptvConsumer {
95 private DataSyncMsg parseContent(String content) { 96 private DataSyncMsg parseContent(String content) {
96 DataSyncMsg dataSyncMsg = JSONUtil.parseMsg2Object(content,DataSyncMsg.class); 97 DataSyncMsg dataSyncMsg = JSONUtil.parseMsg2Object(content,DataSyncMsg.class);
97 Assert.notNull(dataSyncMsg,"ERROR -->> operationConsumer -->> parseContent -->> 【dataSyncMsg】 not be null !!"); 98 Assert.notNull(dataSyncMsg,"ERROR -->> operationConsumer -->> parseContent -->> 【dataSyncMsg】 not be null !!");
98 DataSyncMsg.MsgData msgData = dataSyncMsg.getMsg(); 99 String msgDataStr = dataSyncMsg.getMsgData();
100 DataSyncMsg.MsgData msgData = null;
101 if (StringUtils.isNotBlank(msgDataStr)) {
102 msgData = JSONObject.parseObject(msgDataStr, DataSyncMsg.MsgData.class);
103 }else {
104 return null;
105 }
99 Long memberId = msgData.getMemberId(); 106 Long memberId = msgData.getMemberId();
100 String memberCode = msgData.getMemberCode(); 107 String memberCode = msgData.getMemberCode();
101 if (Objects.nonNull(memberId) && StringUtils.isBlank(memberCode)) { 108 if (Objects.nonNull(memberId) && StringUtils.isBlank(memberCode)) {
...@@ -117,7 +124,13 @@ public class UcGatewayIptv2IptvConsumer { ...@@ -117,7 +124,13 @@ public class UcGatewayIptv2IptvConsumer {
117 } 124 }
118 } 125 }
119 } 126 }
120 Assert.notNull(msgData,"ERROR -->> operationConsumer -->> parseContent -->> 【msgData】 not be null !!"); 127
128 if(Objects.isNull(msgData.getMemberCode()) && Objects.isNull(msgData.getMemberId())) {
129 log.error("会员信息不存在，msgData =>> {}", msgData);
130 return null;
131 }
132
133 dataSyncMsg.setMsgData(JSONObject.toJSONString(msgData));
121 return dataSyncMsg; 134 return dataSyncMsg;
122 } 135 }
123 136
......
...@@ -47,7 +47,7 @@ public class WeiXinEventConsumer { ...@@ -47,7 +47,7 @@ public class WeiXinEventConsumer {
47 @RabbitHandler 47 @RabbitHandler
48 @RabbitListener(queues = "#{rabbitMqSourceConfig.getWechatQueue()}", 48 @RabbitListener(queues = "#{rabbitMqSourceConfig.getWechatQueue()}",
49 containerFactory = "#{rabbitMqSourceConfig.getWechatSource()}", 49 containerFactory = "#{rabbitMqSourceConfig.getWechatSource()}",
50 autoStartup = "#{rabbitMqSourceConfig.getWechatStartUp()}", ackMode = "MANUAL") 50 autoStartup = "#{rabbitMqSourceConfig.getWechatStartUp()}", ackMode = "AUTO")
51 @Transactional 51 @Transactional
52 public void subOrUnSubEvent(Channel channel, Message message, String content) throws IOException { 52 public void subOrUnSubEvent(Channel channel, Message message, String content) throws IOException {
53 try { 53 try {
...@@ -61,23 +61,32 @@ public class WeiXinEventConsumer { ...@@ -61,23 +61,32 @@ public class WeiXinEventConsumer {
61 61
62 String openid = wechatMsg.getString("FromUserName"); 62 String openid = wechatMsg.getString("FromUserName");
63 String msgType = wechatMsg.getString("MsgType"); 63 String msgType = wechatMsg.getString("MsgType");
64
64 if ("event".equals(msgType)) { 65 if ("event".equals(msgType)) {
65 String event = wechatMsg.getString("Event"); 66 String event = wechatMsg.getString("Event");
66 String eventKey = wechatMsg.getString("EventKey"); 67 String eventKey = wechatMsg.getString("EventKey");
67 68
69 log.info("event ==>> {}", event);
70
68 SubscribeBean subscribeBean = new SubscribeBean(); 71 SubscribeBean subscribeBean = new SubscribeBean();
69 subscribeBean.setAppid(appid); 72 subscribeBean.setAppid(appid);
70 subscribeBean.setOpenid(openid); 73 subscribeBean.setOpenid(openid);
71 subscribeBean.setUnionid(unionid); 74 subscribeBean.setUnionid(unionid);
72 subscribeBean.setEventKey(eventKey); 75 subscribeBean.setEventKey(eventKey);
73 76
74 if (event.equals("subscribe")) 77 if (event.equals("subscribe")) {
78 log.info("send subscribe request start");
75 this.restTemplateClient.subscribe(subscribeBean); 79 this.restTemplateClient.subscribe(subscribeBean);
80 log.info("send subscribe request end ");
81 }
76 82
77 if (event.equals("unsubscribe")) 83 if (event.equals("unsubscribe")) {
84 log.info("send unsubscribe request start");
78 this.restTemplateClient.unsubscribe(subscribeBean); 85 this.restTemplateClient.unsubscribe(subscribeBean);
86 log.info("send unsubscribe request end");
87 }
79 88
80 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); 89 // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
81 } 90 }
82 91
83 } catch (Exception e) { 92 } catch (Exception e) {
......
...@@ -26,7 +26,7 @@ public class DataSyncMsg implements Serializable { ...@@ -26,7 +26,7 @@ public class DataSyncMsg implements Serializable {
26 // 发送时间 26 // 发送时间
27 private LocalDateTime time; 27 private LocalDateTime time;
28 // 消息体 28 // 消息体
29 private MsgData msg; 29 private String msgData;
30 30
31 /** 31 /**
32 * 消息体 32 * 消息体
......
...@@ -97,7 +97,9 @@ public class RestTemplateClient { ...@@ -97,7 +97,9 @@ public class RestTemplateClient {
97 97
98 HashMap<Object, Object> objectObjectHashMap = new HashMap<>(); 98 HashMap<Object, Object> objectObjectHashMap = new HashMap<>();
99 objectObjectHashMap.put("content", content); 99 objectObjectHashMap.put("content", content);
100 log.info("reobjectObjectHashMap ===>> [{}]",objectObjectHashMap);
100 restTemplate.postForEntity(url, objectObjectHashMap, String.class); 101 restTemplate.postForEntity(url, objectObjectHashMap, String.class);
102 log.info("unsubscribe ===>> success");
101 /*ResponseEntity<String> responseEntity = restTemplate.postForEntity(url, subscribeBean, String.class); 103 /*ResponseEntity<String> responseEntity = restTemplate.postForEntity(url, subscribeBean, String.class);
102 String entityBody = ""; 104 String entityBody = "";
103 if (responseEntity.getStatusCode().is2xxSuccessful()) { 105 if (responseEntity.getStatusCode().is2xxSuccessful()) {
...@@ -113,8 +115,9 @@ public class RestTemplateClient { ...@@ -113,8 +115,9 @@ public class RestTemplateClient {
113 115
114 HashMap<String, String> objectObjectHashMap = new HashMap<>(); 116 HashMap<String, String> objectObjectHashMap = new HashMap<>();
115 objectObjectHashMap.put("content", content); 117 objectObjectHashMap.put("content", content);
116 log.info("objectObjectHashMap ===>> [{}]",objectObjectHashMap); 118 log.info("reobjectObjectHashMap ===>> [{}]",objectObjectHashMap);
117 restTemplate.postForEntity(url, objectObjectHashMap, String.class); 119 restTemplate.postForEntity(url, objectObjectHashMap, String.class);
120 log.info("send subscribe request ===>> success");
118 /*ResponseEntity<String> responseEntity = restTemplate.postForEntity(url, subscribeBean, String.class); 121 /*ResponseEntity<String> responseEntity = restTemplate.postForEntity(url, subscribeBean, String.class);
119 String entityBody = ""; 122 String entityBody = "";
120 if (responseEntity.getStatusCode().is2xxSuccessful()) { 123 if (responseEntity.getStatusCode().is2xxSuccessful()) {
......