Skip to content

张云鹏 / uc-consumer

Go to a project
Toggle navigation
Toggle navigation pinning
Commit 7bee7486 7bee74860b6093482769176699e11a84a47ee1fd by xianghan
Options

1.优化

1 parent 32ee9bbe
Inline Side-by-side
Showing 6 changed files with 28 additions and 135 deletions
...@@ -18,24 +18,6 @@ import org.springframework.context.annotation.Primary; ...@@ -18,24 +18,6 @@ import org.springframework.context.annotation.Primary;
18 @Configuration 18 @Configuration
19 public class RabbitMqConfig { 19 public class RabbitMqConfig {
20 20
21 //////////////////////////////////////////////////////// OMO 小屏///////////////////////////////////////////////////////////////
22
23 /** 获取带参二维码(大屏->小屏) */
24 public static final String GET_QR_CODE_QUEUE = "queue.qrCode.get";
25
26 /** 删除全部收藏队列(大屏->小屏) */
27 public static final String COLLECTION_DELETE_ALL_QUEUE = "queue.collection.deleteall";
28
29 /** 添加收藏队列(大屏->小屏) */
30 public static final String COLLECTION_ADD_QUEUE = "queue.collection.add";
31
32 /** 删除收藏队列(大屏->小屏) */
33 public static final String COLLECTION_DELETE_QUEUE = "queue.collection.delete";
34
35 /** 微信侧 公众号关注与取消关注 */
36 public static final String WEIXIN_SUBORUNSUB_QUEUE = "weixin.subOrUnSub.queue";
37
38
39 @Value("${mutil-mq.service.host}") 21 @Value("${mutil-mq.service.host}")
40 private String serviceHost; 22 private String serviceHost;
41 @Value("${mutil-mq.service.port}") 23 @Value("${mutil-mq.service.port}")
...@@ -205,4 +187,18 @@ public class RabbitMqConfig { ...@@ -205,4 +187,18 @@ public class RabbitMqConfig {
205 .to(eventBusExchange) 187 .to(eventBusExchange)
206 .with(UC_EVENTBUS_KEY); 188 .with(UC_EVENTBUS_KEY);
207 } 189 }
190
191
192 /**************************************************跨屏数据*************************************************************/
193 /** 删除全部收藏队列(大屏->小屏) */
194 public static final String COLLECTION_DELETE_ALL_QUEUE = "queue.collection.deleteall";
195
196 /** 添加收藏队列(大屏->小屏) */
197 public static final String COLLECTION_ADD_QUEUE = "queue.collection.add";
198
199 /** 删除收藏队列(大屏->小屏) */
200 public static final String COLLECTION_DELETE_QUEUE = "queue.collection.delete";
201
202 /** 微信侧 公众号关注与取消关注 */
203 public static final String WEIXIN_SUBORUNSUB_QUEUE = "weixin.subOrUnSub.queue";
208 } 204 }
......
...@@ -28,11 +28,11 @@ public class UcEngineManagementEventConsumer { ...@@ -28,11 +28,11 @@ public class UcEngineManagementEventConsumer {
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 = "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 TableOperationMsg tableOperationMsg = this.parseContent(content); 38 TableOperationMsg tableOperationMsg = this.parseContent(content);
......
...@@ -38,9 +38,9 @@ public class UcEventBusConsumer { ...@@ -38,9 +38,9 @@ public class UcEventBusConsumer {
38 * @author Hongyan Wang 38 * @author Hongyan Wang
39 * @date 2021/9/7 11:26 上午 39 * @date 2021/9/7 11:26 上午
40 */ 40 */
41 @RabbitHandler 41 /*@RabbitHandler
42 @RabbitListener(queues = RabbitMqConfig.UC_EVENTBUS_TOPIC, 42 @RabbitListener(queues = RabbitMqConfig.UC_EVENTBUS_TOPIC,
43 containerFactory = "managementRabbitListenerContainerFactory") 43 containerFactory = "managementRabbitListenerContainerFactory")*/
44 public void ucEventConsumer(String content) { 44 public void ucEventConsumer(String content) {
45 log.info(" receive dataSync msg , content is : {} ", content); 45 log.info(" receive dataSync msg , content is : {} ", content);
46 DataSyncMsg dataSyncMsg = this.parseContent(content); 46 DataSyncMsg dataSyncMsg = this.parseContent(content);
......
...@@ -28,11 +28,11 @@ public class UcGatewayEventConsumer { ...@@ -28,11 +28,11 @@ public class UcGatewayEventConsumer {
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 = "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 DataSyncMsg dataSyncMsg = this.parseContent(content); 38 DataSyncMsg dataSyncMsg = this.parseContent(content);
......
...@@ -32,11 +32,11 @@ public class WeiXinEventConsumer { ...@@ -32,11 +32,11 @@ public class WeiXinEventConsumer {
32 * @description 删除用户收藏记录 32 * @description 删除用户收藏记录
33 * @param content 消息内容 33 * @param content 消息内容
34 */ 34 */
35 /*@RabbitHandler 35 @RabbitHandler
36 @RabbitListener(bindings = { 36 @RabbitListener(bindings = {
37 @QueueBinding(value = @Queue(value = RabbitMqConfig.COLLECTION_DELETE_QUEUE), 37 @QueueBinding(value = @Queue(value = RabbitMqConfig.COLLECTION_DELETE_QUEUE),
38 exchange = @Exchange(value = ExchangeTypes.DIRECT))}, 38 exchange = @Exchange(value = ExchangeTypes.DIRECT))},
39 containerFactory = "managementRabbitListenerContainerFactory")*/ 39 containerFactory = "managementRabbitListenerContainerFactory")
40 public void deleteCollection(String content) { 40 public void deleteCollection(String content) {
41 try { 41 try {
42 log.info("receive UserCollection delete message, content {}", content); 42 log.info("receive UserCollection delete message, content {}", content);
...@@ -50,11 +50,11 @@ public class WeiXinEventConsumer { ...@@ -50,11 +50,11 @@ public class WeiXinEventConsumer {
50 * @description 删除全部收藏记录 50 * @description 删除全部收藏记录
51 * @param content 消息内容 51 * @param content 消息内容
52 */ 52 */
53 /* @RabbitHandler 53 @RabbitHandler
54 @RabbitListener(bindings = { 54 @RabbitListener(bindings = {
55 @QueueBinding(value = @Queue(value = RabbitMqConfig.COLLECTION_DELETE_ALL_QUEUE), 55 @QueueBinding(value = @Queue(value = RabbitMqConfig.COLLECTION_DELETE_ALL_QUEUE),
56 exchange = @Exchange(value = ExchangeTypes.DIRECT))}, 56 exchange = @Exchange(value = ExchangeTypes.DIRECT))},
57 containerFactory = "managementRabbitListenerContainerFactory")*/ 57 containerFactory = "managementRabbitListenerContainerFactory")
58 @Transactional 58 @Transactional
59 public void deleteAllCollection(String content) { 59 public void deleteAllCollection(String content) {
60 try { 60 try {
...@@ -66,62 +66,6 @@ public class WeiXinEventConsumer { ...@@ -66,62 +66,6 @@ public class WeiXinEventConsumer {
66 } 66 }
67 67
68 /** 68 /**
69 * 处理带参的二维码事件
70 * @param content 消息内容
71 * @description 获取公众号带参二维码
72 */
73 /*@RabbitHandler
74 @RabbitListener(bindings = {
75 @QueueBinding(value = @Queue(value = RabbitMqConfig.GET_QR_CODE_QUEUE),
76 exchange = @Exchange(value = ExchangeTypes.DIRECT))},
77 containerFactory = "managementRabbitListenerContainerFactory")*/
78 public void getQrCode(String content) {
79 try {
80 log.info("receive get qrCode message, content {}", content);
81 JSONObject jsonObject = JSONObject.parseObject(content);
82 /*String appid = jsonObject.getString("appid");
83 String IPTVappid = jsonObject.getString("IPTVappid");
84 String platformAccount = jsonObject.getString("platformAccount");
85 String sessionId = jsonObject.getString("sessionId");
86 String key = QR_CODE_URL + appid + "_" + platformAccount + "_" + sessionId;
87 String url = (String) redisUtils.get(key);
88 if (StringUtils.isBlank(url)) {
89 Map<String, String> wxInfo = WeixinUtil.getWeixinInfoByAppid(appid);
90 var appType = wxInfo.get("appType");
91 // 订阅号不支持带参二维码，直接返回
92 if (StrUtil.isNotEmpty(appType) && ObjectUtil.equals(appType, WeChatConstants.WX_SUBSCRIPTION)) {
93 log.error("订阅号不支持带参二维码 || {} || {}", appid, content);
94 return;
95 }
96 QrCode qrCode = new QrCode();
97 qrCode.setActionName(WeChatConstants.QR_STR_SCENE);
98 if (StringUtils.isNotBlank(wxInfo.get("qrCodeExpireSeconds"))) {
99 qrCode.setExpireSeconds(Integer.valueOf(wxInfo.get("qrCodeExpireSeconds")));
100 }
101 ActionInfo actionInfo = new ActionInfo();
102 Scene scene = new Scene();
103 scene.setSceneStr(content);
104 actionInfo.setScene(scene);
105 qrCode.setActionInfo(actionInfo);
106 JSONObject jsonQrCode = weixinRequestUtil.getQrCode(wxInfo, qrCode);
107 url = jsonQrCode.getString("url");
108 Integer expireSeconds = jsonQrCode.getInteger("expire_seconds");
109 redisUtils.set(key, url, expireSeconds, TimeUnit.SECONDS);
110 }
111 HashMap<String, Object> map = new HashMap<>();
112 map.put("sessionId", sessionId);
113 map.put("url", url);
114 map.put("appid", appid);
115 map.put("IPTVappid", IPTVappid);
116 map.put("platformAccount", platformAccount);
117 map.put("extraInfo", content);*/
118 restTemplateClient.sendQrCodeMessage(content);
119 } catch (Exception e) {
120 log.error("GetQrCodeConsumer || get qrCode error || {}", e.toString(), e);
121 }
122 }
123
124 /**
125 * 关注和取关事件 69 * 关注和取关事件
126 * eg: 70 * eg:
127 * { 71 * {
...@@ -130,11 +74,11 @@ public class WeiXinEventConsumer { ...@@ -130,11 +74,11 @@ public class WeiXinEventConsumer {
130 * } 74 * }
131 * @param content 75 * @param content
132 */ 76 */
133 /* @RabbitHandler 77 @RabbitHandler
134 @RabbitListener(bindings = { 78 @RabbitListener(bindings = {
135 @QueueBinding(value = @Queue(value = RabbitMqConfig.WEIXIN_SUBORUNSUB_QUEUE), 79 @QueueBinding(value = @Queue(value = RabbitMqConfig.WEIXIN_SUBORUNSUB_QUEUE),
136 exchange = @Exchange(value = ExchangeTypes.DIRECT))}, 80 exchange = @Exchange(value = ExchangeTypes.DIRECT))},
137 containerFactory = "managementRabbitListenerContainerFactory")*/ 81 containerFactory = "managementRabbitListenerContainerFactory")
138 @Transactional 82 @Transactional
139 public void subOrUnSubEvent(String content) { 83 public void subOrUnSubEvent(String content) {
140 try { 84 try {
...@@ -145,7 +89,6 @@ public class WeiXinEventConsumer { ...@@ -145,7 +89,6 @@ public class WeiXinEventConsumer {
145 JSONObject wechatMsg = jsonObject.getJSONObject("allFieldsMap"); 89 JSONObject wechatMsg = jsonObject.getJSONObject("allFieldsMap");
146 String appid = map.getString("mpId"); 90 String appid = map.getString("mpId");
147 String unionid = map.getString("unionid"); 91 String unionid = map.getString("unionid");
148 // Map<String, String> wxInfoMap = WeixinUtil.getWeixinInfoByAppid(appid);
149 92
150 String openid = wechatMsg.getString("FromUserName"); 93 String openid = wechatMsg.getString("FromUserName");
151 String msgType = wechatMsg.getString("MsgType"); 94 String msgType = wechatMsg.getString("MsgType");
...@@ -186,52 +129,6 @@ public class WeiXinEventConsumer { ...@@ -186,52 +129,6 @@ public class WeiXinEventConsumer {
186 public void addCollection(String content) { 129 public void addCollection(String content) {
187 try { 130 try {
188 log.info("receive UserCollection add message, content {}", content); 131 log.info("receive UserCollection add message, content {}", content);
189 JSONObject jsonObject = JSONObject.parseObject(content);
190 String platformAccount = jsonObject.getString("platformAccount");
191 String data = jsonObject.getString("data");
192 if (StringUtils.isBlank(data) || !data.startsWith("[")) {
193 // return;
194 }
195 /*Optional<TvUser> userOptional = tvUserRepository.findByPlatformAccount(platformAccount);
196 if (!userOptional.isPresent()) {
197 return;
198 }
199 Long tvUserId = userOptional.get().getId();
200 List<UserCollectionMq> userCollectionMqList = JSONObject.parseArray(data, UserCollectionMq.class);
201 if (userCollectionMqList == null || userCollectionMqList.isEmpty()) {
202 return;
203 }
204 Map<Long, List<UserCollectionMq>> collect = userCollectionMqList.stream().collect(Collectors.groupingBy(UserCollectionMq::getUserCollectionId));
205 for (Map.Entry<Long, List<UserCollectionMq>> entry : collect.entrySet()) {
206 List<UserCollectionMq> value = entry.getValue();
207 UserCollectionMq userCollectionMq = value.get(0);
208 if (StringUtils.isBlank(userCollectionMq.getName())) {
209 userCollectionMq.setName("DEFAULT");
210 }
211 UserCollection userCollection = userCollectionRepository
212 .findFirstByUserIdAndTypeAndName(tvUserId, userCollectionMq.getType(), userCollectionMq.getName()).orElseGet(UserCollection::new);
213 userCollection.setAppId(userCollectionMq.getAppId())
214 .setUserId(tvUserId)
215 .setName(userCollectionMq.getName())
216 .setType(userCollectionMq.getType())
217 .setCount(userCollection.getCount() == null ? value.size() : userCollection.getCount() + value.size());
218 UserCollection userCollectionSave = userCollectionRepository.save(userCollection);
219 for (UserCollectionMq collectionMq : value) {
220 UserCollectionDetail userCollectionDetail = collectionMq2DetailMapper.toEntity(collectionMq);
221 Optional<UserCollectionDetail> userCollectionDetailOptional = userCollectionDetailRepository
222 .findByDetailIdAndDetailTypeAndUserCollectionId(userCollectionDetail.getDetailId(), userCollectionDetail.getDetailType(), userCollectionSave.getId());
223 //观影记录同一天只存一条记录
224 if (userCollectionDetailOptional.isPresent() &&
225 DateUtil.isSameDay(new Date(userCollectionDetailOptional.get().getCreateTime().getTime()), new Date())) {
226 userCollectionDetail.setId(userCollectionDetailOptional.get().getId());
227 } else {
228 userCollectionDetail.setId(null)
229 .setUserCollectionId(userCollectionSave.getId());
230 }
231 userCollectionDetailRepository.save(userCollectionDetail);
232 }
233 }*/
234
235 this.restTemplateClient.addCollection(content); 132 this.restTemplateClient.addCollection(content);
236 } catch (Exception e) { 133 } catch (Exception e) {
237 log.error("CollectionAddConsumer || UserCollection add error || {}", e.toString(), e); 134 log.error("CollectionAddConsumer || UserCollection add error || {}", e.toString(), e);
......
...@@ -119,9 +119,9 @@ mutil-mq: ...@@ -119,9 +119,9 @@ mutil-mq:
119 # 服务属性 119 # 服务属性
120 service: 120 service:
121 #平台类型 service: 服务侧 management: 管理侧 121 #平台类型 service: 服务侧 management: 管理侧
122 platform: service 122 platform: management
123 # 服务域 mobile:小屏侧 vis:大屏侧 123 # 服务域 mobile:小屏侧 vis:大屏侧
124 type: vis 124 type:
125 125
126 # uc-engine服务地址 126 # uc-engine服务地址
127 api: 127 api:
......