Skip to content

向汉 / uc-consumer

Go to a project
Toggle navigation
Toggle navigation pinning
Commit b5fbd670 b5fbd6701730057cd8401280e7927030aaa63f3b by xianghan
Options

1.update

1 parent 64fe16cc
Inline Side-by-side
Showing 6 changed files with 130 additions and 22 deletions
...@@ -35,7 +35,7 @@ public class UcEngineIptv2ManagementConsumer { ...@@ -35,7 +35,7 @@ public class UcEngineIptv2ManagementConsumer {
35 @RabbitListener(bindings = { 35 @RabbitListener(bindings = {
36 @QueueBinding(value = @Queue(value = RabbitMqConfig.ENGINE_TO_MANAGEMENT_DIRECT), 36 @QueueBinding(value = @Queue(value = RabbitMqConfig.ENGINE_TO_MANAGEMENT_DIRECT),
37 exchange = @Exchange(value = ExchangeTypes.DIRECT)) 37 exchange = @Exchange(value = ExchangeTypes.DIRECT))
38 }, containerFactory = "serviceRabbitListenerContainerFactory")*/ 38 }, containerFactory = "managementRabbitListenerContainerFactory")*/
39 public void ucEventConsumer(String content) { 39 public void ucEventConsumer(String content) {
40 log.info(" receive dataSync msg , content is : {} ", content); 40 log.info(" receive dataSync msg , content is : {} ", content);
41 TableOperationMsg tableOperationMsg = this.parseContent(content); 41 TableOperationMsg tableOperationMsg = this.parseContent(content);
......
...@@ -28,11 +28,11 @@ public class UcEngineManagement2IptvConsumer { ...@@ -28,11 +28,11 @@ public class UcEngineManagement2IptvConsumer {
28 * @author Hongyan Wang 28 * @author Hongyan Wang
29 * @date 2021/9/7 11:26 上午 29 * @date 2021/9/7 11:26 上午
30 */ 30 */
31 /*@RabbitHandler 31 @RabbitHandler
32 @RabbitListener(bindings = { 32 @RabbitListener(bindings = {
33 @QueueBinding(value = @Queue(value = RabbitMqConfig.ENGINE_TO_SERVICE_DIRECT), 33 @QueueBinding(value = @Queue(value = RabbitMqConfig.ENGINE_TO_SERVICE_DIRECT),
34 exchange = @Exchange(value = ExchangeTypes.DIRECT)) 34 exchange = @Exchange(value = ExchangeTypes.DIRECT))
35 }, containerFactory = "serviceRabbitListenerContainerFactory")*/ 35 }, containerFactory = "managementRabbitListenerContainerFactory")
36 public void ucEventConsumer(String content) { 36 public void ucEventConsumer(String content) {
37 log.info(" receive dataSync msg , content is : {} ", content); 37 log.info(" receive dataSync msg , content is : {} ", content);
38 TableOperationMsg tableOperationMsg = this.parseContent(content); 38 TableOperationMsg tableOperationMsg = this.parseContent(content);
......
...@@ -11,15 +11,25 @@ import com.topdraw.exception.BadRequestException; ...@@ -11,15 +11,25 @@ import com.topdraw.exception.BadRequestException;
11 import com.topdraw.exception.EntityNotFoundException; 11 import com.topdraw.exception.EntityNotFoundException;
12 import com.topdraw.mq.domain.DataSyncMsg; 12 import com.topdraw.mq.domain.DataSyncMsg;
13 import com.topdraw.resttemplate.RestTemplateClient; 13 import com.topdraw.resttemplate.RestTemplateClient;
14 import com.topdraw.util.DateUtil;
14 import com.topdraw.util.JSONUtil; 15 import com.topdraw.util.JSONUtil;
16 import com.topdraw.util.TimestampUtil;
17 import com.topdraw.utils.RedisUtils;
15 import lombok.Data; 18 import lombok.Data;
16 import lombok.extern.slf4j.Slf4j; 19 import lombok.extern.slf4j.Slf4j;
20 import org.apache.commons.collections4.MapUtils;
17 import org.apache.commons.lang3.StringUtils; 21 import org.apache.commons.lang3.StringUtils;
22 import org.apache.commons.lang3.time.DateFormatUtils;
23 import org.apache.commons.lang3.time.DateUtils;
18 import org.springframework.amqp.rabbit.annotation.*; 24 import org.springframework.amqp.rabbit.annotation.*;
19 import org.springframework.beans.factory.annotation.Autowired; 25 import org.springframework.beans.factory.annotation.Autowired;
20 import org.springframework.stereotype.Component; 26 import org.springframework.stereotype.Component;
21 27
22 import java.util.Objects; 28 import java.text.DateFormat;
29 import java.text.ParseException;
30 import java.text.SimpleDateFormat;
31 import java.time.LocalDateTime;
32 import java.util.*;
23 33
24 @Component 34 @Component
25 @Slf4j 35 @Slf4j
...@@ -32,6 +42,12 @@ public class UcEventBusIptv2ManagementUcEngine { ...@@ -32,6 +42,12 @@ public class UcEventBusIptv2ManagementUcEngine {
32 @Autowired 42 @Autowired
33 private MemberService memberService; 43 private MemberService memberService;
34 44
45 @Autowired
46 private RedisUtils redisUtils;
47
48
49 private static final Integer PLAY_30 = 30;
50
35 /** 51 /**
36 * 事件 52 * 事件
37 * @param content 53 * @param content
...@@ -39,22 +55,25 @@ public class UcEventBusIptv2ManagementUcEngine { ...@@ -39,22 +55,25 @@ public class UcEventBusIptv2ManagementUcEngine {
39 * @author Hongyan Wang 55 * @author Hongyan Wang
40 * @date 2021/9/7 11:26 上午 56 * @date 2021/9/7 11:26 上午
41 */ 57 */
42 /*@RabbitHandler 58 @RabbitHandler
43 @RabbitListener(queues = RabbitMqConfig.UC_EVENTBUS_QUEUE, 59 @RabbitListener(queues = RabbitMqConfig.UC_EVENTBUS_QUEUE,
44 containerFactory = "managementRabbitListenerContainerFactory")*/ 60 containerFactory = "serviceRabbitListenerContainerFactory")
45 public void ucEventConsumer(String content) { 61 public void ucEventConsumer(String content) throws ParseException {
46 log.info(" receive dataSync msg , content is : {} ", content); 62 log.info(" receive dataSync msg , content is : {} ", content);
47 DataSyncMsg dataSyncMsg = this.parseContent(content); 63 DataSyncMsg dataSyncMsg = this.parseContent(content);
64 if (Objects.nonNull(dataSyncMsg)) {
48 this.taskDeal(dataSyncMsg); 65 this.taskDeal(dataSyncMsg);
66 }
49 log.info("ucEventConsumer ====>>>> end"); 67 log.info("ucEventConsumer ====>>>> end");
50 } 68 }
51 69
70
52 /** 71 /**
53 * 数据解析 72 * 数据解析
54 * @param content 73 * @param content
55 * @return 74 * @return
56 */ 75 */
57 private DataSyncMsg parseContent(String content) { 76 private DataSyncMsg parseContent(String content) throws ParseException {
58 77
59 CommonMsg commonMsg = JSONUtil.parseMsg2Object(content, CommonMsg.class); 78 CommonMsg commonMsg = JSONUtil.parseMsg2Object(content, CommonMsg.class);
60 79
...@@ -67,6 +86,7 @@ public class UcEventBusIptv2ManagementUcEngine { ...@@ -67,6 +86,7 @@ public class UcEventBusIptv2ManagementUcEngine {
67 PlayContent.MsgData msgData = playContent.getMsgData(); 86 PlayContent.MsgData msgData = playContent.getMsgData();
68 if (Objects.nonNull(msgData)) { 87 if (Objects.nonNull(msgData)) {
69 String time = playContent.getTime(); 88 String time = playContent.getTime();
89 String formatDate = DateUtil.formatDate(time);
70 Integer deviceType = playContent.getDeviceType(); 90 Integer deviceType = playContent.getDeviceType();
71 String platformAccount = msgData.getPlatformAccount(); 91 String platformAccount = msgData.getPlatformAccount();
72 92
...@@ -74,14 +94,95 @@ public class UcEventBusIptv2ManagementUcEngine { ...@@ -74,14 +94,95 @@ public class UcEventBusIptv2ManagementUcEngine {
74 Long mediaId = msgData.getMediaId(); 94 Long mediaId = msgData.getMediaId();
75 String mediaName = msgData.getMediaName(); 95 String mediaName = msgData.getMediaName();
76 Integer playDuration = msgData.getPlayDuration(); 96 Integer playDuration = msgData.getPlayDuration();
97 log.info("playDuration ==>> {}", playDuration);
77 98
78 DataSyncMsg dataSyncMsg = new DataSyncMsg(); 99 DataSyncMsg dataSyncMsg = new DataSyncMsg();
79 dataSyncMsg.setEventType(evt); 100 dataSyncMsg.setEventType(evt);
80 DataSyncMsg.MsgData msgData1 = new DataSyncMsg.MsgData(); 101 DataSyncMsg.MsgData msgData1 = new DataSyncMsg.MsgData();
81 102
82 if (StringUtils.isNotBlank(platformAccount)) { 103 if (StringUtils.isNotBlank(platformAccount)) {
104
83 UserTvDTO userTvDTO = this.userTvService.findByPlatformAccount(platformAccount); 105 UserTvDTO userTvDTO = this.userTvService.findByPlatformAccount(platformAccount);
106
107 if(Objects.nonNull(userTvDTO)) {
108
109 String key = platformAccount+"|"+formatDate;
110 Map<Object, Object> hmget =
111 this.redisUtils.hmget(key);
112
113 Integer playDurationValueTotal = 0;
114 int maxSize = 1;
115 if (MapUtils.isNotEmpty(hmget)) {
116 Set<Object> objects = hmget.keySet();
117 for (Object key_ : objects) {
118
119 if (Objects.nonNull(key_)) {
120
121 if (key_.toString().equalsIgnoreCase("total")) {
122 Integer playDurationValueTotal_ = Integer.valueOf(hmget.get(key_).toString());
123 if (playDurationValueTotal_ >= 30) {
124 maxSize = objects.size();
125 Integer maxTotal = maxSize+1;
126 Integer playDurationValue = Integer.valueOf(hmget.get(key_).toString());
127 playDurationValueTotal = playDurationValue+playDuration;
128 Map<String, Object> map = new HashMap<>();
129 map.put(String.valueOf(maxTotal), playDuration);
130 map.put("total", playDurationValueTotal);
131 this.redisUtils.hmset(key, map, 172800);
132 return null;
133 }
134 }
135
136 maxSize = objects.size();
137
138 Integer playDurationValue = Integer.valueOf(hmget.get("total").toString());
139 playDurationValueTotal = playDurationValue+playDuration;
140
141 Map<String, Object> map = new HashMap<>();
142 map.put(String.valueOf(maxSize+1), playDuration);
143 map.put("total", playDurationValueTotal);
144 this.redisUtils.hmset(key, map, 172800);
145
146 }
147
148 }
149
150 } else {
151 playDurationValueTotal = playDuration;
152 Map<String, Object> map = new HashMap<>();
153 map.put("total", playDurationValueTotal);
154 map.put("1", playDuration);
155 this.redisUtils.hmset(key, map, 172800);
156
157 }
158
159 DataSyncMsg dataSyncMsg1 = null;
160 if (playDurationValueTotal >= PLAY_30) {
161 log.info("playDurationValueTotal ===>>> {}",playDurationValueTotal);
162 log.info("===>> start dealTask");
163 dataSyncMsg1 = getDataSyncMsg(time, deviceType, mediaCode, mediaId, mediaName, playDuration,
164 dataSyncMsg, msgData1, userTvDTO);
165 }
166
167 return dataSyncMsg1;
168
169 }
170
171 }
172
173 }
174
175 System.out.println(playContent);
176 break;
177
178 }
179
180 return null;
181 }
182
183 private DataSyncMsg getDataSyncMsg(String time, Integer deviceType, String mediaCode, Long mediaId, String mediaName, Integer playDuration, DataSyncMsg dataSyncMsg, DataSyncMsg.MsgData msgData1, UserTvDTO userTvDTO) {
84 String priorityMemberCode = userTvDTO.getPriorityMemberCode(); 184 String priorityMemberCode = userTvDTO.getPriorityMemberCode();
185 log.info("priorityMemberCode ==>> {}", priorityMemberCode);
85 String memberCode = ""; 186 String memberCode = "";
86 if (StringUtils.isNotBlank(priorityMemberCode)) { 187 if (StringUtils.isNotBlank(priorityMemberCode)) {
87 memberCode = priorityMemberCode; 188 memberCode = priorityMemberCode;
...@@ -89,17 +190,18 @@ public class UcEventBusIptv2ManagementUcEngine { ...@@ -89,17 +190,18 @@ public class UcEventBusIptv2ManagementUcEngine {
89 memberCode = this.memberService.findById(userTvDTO.getMemberId()).getCode(); 190 memberCode = this.memberService.findById(userTvDTO.getMemberId()).getCode();
90 } 191 }
91 192
193 log.info("memberCode ==>> {}", priorityMemberCode);
92 if (StringUtils.isBlank(memberCode)) 194 if (StringUtils.isBlank(memberCode))
93 throw new EntityNotFoundException(MemberDTO.class, "memberCode", "memberCode is null"); 195 throw new EntityNotFoundException(MemberDTO.class, "memberCode", "memberCode is null");
94 196
95 msgData1.setMemberCode(memberCode); 197 msgData1.setMemberCode(memberCode);
96 }
97 198
98 msgData1.setEvent(8); 199 msgData1.setEvent(8);
99 msgData1.setDeviceType(deviceType); 200 msgData1.setDeviceType(deviceType);
100 msgData1.setMediaId(mediaId); 201 msgData1.setMediaId(mediaId);
101 202
102 JSONObject param = new JSONObject(); 203 JSONObject param = new JSONObject();
204 // 增量
103 param.put("playDuration", playDuration); 205 param.put("playDuration", playDuration);
104 msgData1.setParam(JSON.toJSONString(param)); 206 msgData1.setParam(JSON.toJSONString(param));
105 JSONObject description = new JSONObject(); 207 JSONObject description = new JSONObject();
...@@ -109,19 +211,11 @@ public class UcEventBusIptv2ManagementUcEngine { ...@@ -109,19 +211,11 @@ public class UcEventBusIptv2ManagementUcEngine {
109 description.put("mediaCode", mediaCode); 211 description.put("mediaCode", mediaCode);
110 description.put("time", time); 212 description.put("time", time);
111 msgData1.setDescription(JSON.toJSONString(description)); 213 msgData1.setDescription(JSON.toJSONString(description));
112
113 dataSyncMsg.setMsg(msgData1); 214 dataSyncMsg.setMsg(msgData1);
215 log.info("dataSyncMsg ==>> {}", dataSyncMsg);
114 return dataSyncMsg; 216 return dataSyncMsg;
115 } 217 }
116 218
117 System.out.println(playContent);
118 break;
119
120 }
121
122 return null;
123 }
124
125 /** 219 /**
126 * 任务处理 220 * 任务处理
127 * @param dataSyncMsg 221 * @param dataSyncMsg
......
...@@ -28,11 +28,11 @@ public class UcGatewayIptv2IptvConsumer { ...@@ -28,11 +28,11 @@ public class UcGatewayIptv2IptvConsumer {
28 * @author Hongyan Wang 28 * @author Hongyan Wang
29 * @date 2021/9/7 11:26 上午 29 * @date 2021/9/7 11:26 上午
30 */ 30 */
31 /*@RabbitHandler 31 @RabbitHandler
32 @RabbitListener(bindings = { 32 @RabbitListener(bindings = {
33 @QueueBinding(value = @Queue(value = RabbitMqConfig.GATEWAY_TO_SERVICE_DIRECT), 33 @QueueBinding(value = @Queue(value = RabbitMqConfig.GATEWAY_TO_SERVICE_DIRECT),
34 exchange = @Exchange(value = ExchangeTypes.DIRECT)) 34 exchange = @Exchange(value = ExchangeTypes.DIRECT))
35 }, containerFactory = "managementRabbitListenerContainerFactory")*/ 35 }, containerFactory = "serviceRabbitListenerContainerFactory")
36 public void ucEventConsumer(String content) { 36 public void ucEventConsumer(String content) {
37 log.info(" receive dataSync msg , content is : {} ", content); 37 log.info(" receive dataSync msg , content is : {} ", content);
38 DataSyncMsg dataSyncMsg = this.parseContent(content); 38 DataSyncMsg dataSyncMsg = this.parseContent(content);
......
...@@ -74,11 +74,11 @@ public class WeiXinEventConsumer { ...@@ -74,11 +74,11 @@ public class WeiXinEventConsumer {
74 * } 74 * }
75 * @param content 75 * @param content
76 */ 76 */
77 @RabbitHandler 77 /* @RabbitHandler
78 @RabbitListener(bindings = { 78 @RabbitListener(bindings = {
79 @QueueBinding(value = @Queue(value = RabbitMqConfig.WEIXIN_SUBORUNSUB_QUEUE), 79 @QueueBinding(value = @Queue(value = RabbitMqConfig.WEIXIN_SUBORUNSUB_QUEUE),
80 exchange = @Exchange(value = ExchangeTypes.DIRECT))}, 80 exchange = @Exchange(value = ExchangeTypes.DIRECT))},
81 containerFactory = "managementRabbitListenerContainerFactory") 81 containerFactory = "managementRabbitListenerContainerFactory")*/
82 @Transactional 82 @Transactional
83 public void subOrUnSubEvent(String content) { 83 public void subOrUnSubEvent(String content) {
84 try { 84 try {
......
1 package com.topdraw.util; 1 package com.topdraw.util;
2 2
3 import java.text.ParseException;
4 import java.text.SimpleDateFormat;
3 import java.util.Date; 5 import java.util.Date;
4 6
5 public class DateUtil { 7 public class DateUtil {
...@@ -12,5 +14,17 @@ public class DateUtil { ...@@ -12,5 +14,17 @@ public class DateUtil {
12 return System.currentTimeMillis(); 14 return System.currentTimeMillis();
13 } 15 }
14 16
17 public static String formatDate(String time){
18 SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");
19 try {
20 //使用SimpleDateFormat的parse()方法生成Date
21 Date date = sf.parse(time);
22 String format = sf.format(date);
23 return format;
24 } catch (ParseException e) {
25 e.printStackTrace();
26 }
15 27
28 return null;
29 }
16 } 30 }
......