Skip to content

向汉 / uc-consumer

Go to a project
Toggle navigation
Toggle navigation pinning
Commit 1a39d208 1a39d2088c557b8a799be958bad87ae0262c8356 by xianghan
Options

1.优化mq配置，添加默认数据源和默认队列

1 parent aae2399c
Inline Side-by-side
Showing 2 changed files with 162 additions and 110 deletions
...@@ -18,6 +18,7 @@ import org.springframework.context.annotation.Primary; ...@@ -18,6 +18,7 @@ import org.springframework.context.annotation.Primary;
18 18
19 import javax.naming.ConfigurationException; 19 import javax.naming.ConfigurationException;
20 import java.util.Map; 20 import java.util.Map;
21 import java.util.Objects;
21 22
22 @Data 23 @Data
23 @Configuration 24 @Configuration
...@@ -166,32 +167,38 @@ public class RabbitMqSourceConfig { ...@@ -166,32 +167,38 @@ public class RabbitMqSourceConfig {
166 private Map<String, String> ucgEventInfo; 167 private Map<String, String> ucgEventInfo;
167 168
168 public String getUcgEventQueue(){ 169 public String getUcgEventQueue(){
169 if (MapUtils.isNotEmpty(ucgEventInfo)) { 170 if (Objects.nonNull(ucgEventInfo)) {
170 String queue = ucgEventInfo.get("queue"); 171 if (MapUtils.isNotEmpty(ucgEventInfo)) {
171 return queue; 172 String queue = ucgEventInfo.get("queue");
173 return queue;
174 }
172 } 175 }
173 176
174 return EVENT_QUEUE; 177 return EVENT_QUEUE;
175 } 178 }
176 179
177 public String getUcgEventSource(){ 180 public String getUcgEventSource(){
178 if (MapUtils.isNotEmpty(ucgEventInfo)) { 181 if (Objects.nonNull(ucgEventInfo)) {
179 String source = ucgEventInfo.get("active"); 182 if (MapUtils.isNotEmpty(ucgEventInfo)) {
180 if (StringUtils.isNotBlank(source)) { 183 String source = ucgEventInfo.get("active");
181 return this.chargeSource(source); 184 if (StringUtils.isNotBlank(source)) {
182 } else { 185 return this.chargeSource(source);
183 return ""; 186 } else {
187 return SERVICE_;
188 }
184 } 189 }
185 } 190 }
186 191
187 return null; 192 return SERVICE_;
188 } 193 }
189 194
190 public String getUcgEventStartUp(){ 195 public String getUcgEventStartUp(){
191 if (MapUtils.isNotEmpty(ucgEventInfo)) { 196 if (Objects.nonNull(ucgEventInfo)) {
192 String source = ucgEventInfo.get("active"); 197 if (MapUtils.isNotEmpty(ucgEventInfo)) {
193 if (StringUtils.isNotBlank(source)) { 198 String source = ucgEventInfo.get("active");
194 return "true"; 199 if (StringUtils.isNotBlank(source)) {
200 return "true";
201 }
195 } 202 }
196 } 203 }
197 204
...@@ -209,32 +216,38 @@ public class RabbitMqSourceConfig { ...@@ -209,32 +216,38 @@ public class RabbitMqSourceConfig {
209 private Map<String, String> ucgIptvCollectionInfo; 216 private Map<String, String> ucgIptvCollectionInfo;
210 217
211 public String getUcgCollectionQueue(){ 218 public String getUcgCollectionQueue(){
212 if (MapUtils.isNotEmpty(ucgIptvCollectionInfo)) { 219 if (Objects.nonNull(ucgIptvCollectionInfo)) {
213 String queue = ucgIptvCollectionInfo.get("queue"); 220 if (MapUtils.isNotEmpty(ucgIptvCollectionInfo)) {
214 return queue; 221 String queue = ucgIptvCollectionInfo.get("queue");
222 return queue;
223 }
215 } 224 }
216 225
217 return COLLECTION_QUEUE; 226 return COLLECTION_QUEUE;
218 } 227 }
219 228
220 public String getUcgCollectionSource(){ 229 public String getUcgCollectionSource(){
221 if (MapUtils.isNotEmpty(ucgIptvCollectionInfo)) { 230 if (Objects.nonNull(ucgIptvCollectionInfo)) {
222 String source = ucgIptvCollectionInfo.get("active"); 231 if (MapUtils.isNotEmpty(ucgIptvCollectionInfo)) {
223 if (StringUtils.isNotBlank(source)) { 232 String source = ucgIptvCollectionInfo.get("active");
224 return this.chargeSource(source); 233 if (StringUtils.isNotBlank(source)) {
225 } else { 234 return this.chargeSource(source);
226 return ""; 235 } else {
236 return SERVICE_;
237 }
227 } 238 }
228 } 239 }
229 240
230 return null; 241 return SERVICE_;
231 } 242 }
232 243
233 public String getUcgCollectionStartUp(){ 244 public String getUcgCollectionStartUp(){
234 if (MapUtils.isNotEmpty(ucgIptvCollectionInfo)) { 245 if (Objects.nonNull(ucgIptvCollectionInfo)) {
235 String source = ucgIptvCollectionInfo.get("active"); 246 if (MapUtils.isNotEmpty(ucgIptvCollectionInfo)) {
236 if (StringUtils.isNotBlank(source)) { 247 String source = ucgIptvCollectionInfo.get("active");
237 return "true"; 248 if (StringUtils.isNotBlank(source)) {
249 return "true";
250 }
238 } 251 }
239 } 252 }
240 253
...@@ -248,32 +261,38 @@ public class RabbitMqSourceConfig { ...@@ -248,32 +261,38 @@ public class RabbitMqSourceConfig {
248 private Map<String, String> viewRecordInfo; 261 private Map<String, String> viewRecordInfo;
249 262
250 public String getViewRecordQueue(){ 263 public String getViewRecordQueue(){
251 if (MapUtils.isNotEmpty(viewRecordInfo)) { 264 if (Objects.nonNull(viewRecordInfo)) {
252 String queue = viewRecordInfo.get("queue"); 265 if (MapUtils.isNotEmpty(viewRecordInfo)) {
253 return queue; 266 String queue = viewRecordInfo.get("queue");
267 return queue;
268 }
254 } 269 }
255 270
256 return VIEW_RECORD_QUEUE; 271 return VIEW_RECORD_QUEUE;
257 } 272 }
258 273
259 public String getViewRecordSource(){ 274 public String getViewRecordSource(){
260 if (MapUtils.isNotEmpty(viewRecordInfo)) { 275 if (Objects.nonNull(viewRecordInfo)) {
261 String source = viewRecordInfo.get("active"); 276 if (MapUtils.isNotEmpty(viewRecordInfo)) {
262 if (StringUtils.isNotBlank(source)) { 277 String source = viewRecordInfo.get("active");
263 return this.chargeSource(source); 278 if (StringUtils.isNotBlank(source)) {
264 } else { 279 return this.chargeSource(source);
265 return ""; 280 } else {
281 return SERVICE_;
282 }
266 } 283 }
267 } 284 }
268 285
269 return null; 286 return SERVICE_;
270 } 287 }
271 288
272 public String getViewRecordStartUp(){ 289 public String getViewRecordStartUp(){
273 if (MapUtils.isNotEmpty(viewRecordInfo)) { 290 if (Objects.nonNull(viewRecordInfo)) {
274 String source = viewRecordInfo.get("active"); 291 if (MapUtils.isNotEmpty(viewRecordInfo)) {
275 if (StringUtils.isNotBlank(source)) { 292 String source = viewRecordInfo.get("active");
276 return "true"; 293 if (StringUtils.isNotBlank(source)) {
294 return "true";
295 }
277 } 296 }
278 } 297 }
279 298
...@@ -289,32 +308,38 @@ public class RabbitMqSourceConfig { ...@@ -289,32 +308,38 @@ public class RabbitMqSourceConfig {
289 private Map<String, String> uceInfo; 308 private Map<String, String> uceInfo;
290 309
291 public String getUceQueue(){ 310 public String getUceQueue(){
292 if (MapUtils.isNotEmpty(uceInfo)) { 311 if (Objects.nonNull(uceInfo)) {
293 String queue = uceInfo.get("queue"); 312 if (MapUtils.isNotEmpty(uceInfo)) {
294 return queue; 313 String queue = uceInfo.get("queue");
314 return queue;
315 }
295 } 316 }
296 317
297 return UCE_QUEUE; 318 return UCE_QUEUE;
298 } 319 }
299 320
300 public String getUceSource(){ 321 public String getUceSource(){
301 if (MapUtils.isNotEmpty(uceInfo)) { 322 if (Objects.nonNull(uceInfo)) {
302 String source = uceInfo.get("active"); 323 if (MapUtils.isNotEmpty(uceInfo)) {
303 if (StringUtils.isNotBlank(source)) { 324 String source = uceInfo.get("active");
304 return this.chargeSource(source); 325 if (StringUtils.isNotBlank(source)) {
305 } else { 326 return this.chargeSource(source);
306 return ""; 327 } else {
328 return MANAGEMENT_;
329 }
307 } 330 }
308 } 331 }
309 332
310 return null; 333 return MANAGEMENT_;
311 } 334 }
312 335
313 public String getUceStartUp(){ 336 public String getUceStartUp(){
314 if (MapUtils.isNotEmpty(uceInfo)) { 337 if (Objects.nonNull(uceInfo)) {
315 String source = uceInfo.get("active"); 338 if (MapUtils.isNotEmpty(uceInfo)) {
316 if (StringUtils.isNotBlank(source)) { 339 String source = uceInfo.get("active");
317 return "true"; 340 if (StringUtils.isNotBlank(source)) {
341 return "true";
342 }
318 } 343 }
319 } 344 }
320 345
...@@ -330,35 +355,39 @@ public class RabbitMqSourceConfig { ...@@ -330,35 +355,39 @@ public class RabbitMqSourceConfig {
330 private Map<String, String> eventBusInfo; 355 private Map<String, String> eventBusInfo;
331 356
332 public String getEventBusQueue(){ 357 public String getEventBusQueue(){
333 if (MapUtils.isNotEmpty(eventBusInfo)) { 358 if (Objects.nonNull(eventBusInfo)) {
334 String queue = eventBusInfo.get("queue"); 359 if (MapUtils.isNotEmpty(eventBusInfo)) {
335 return queue; 360 String queue = eventBusInfo.get("queue");
361 return queue;
362 }
336 } 363 }
337
338 return UC_EVENTBUS_QUEUE; 364 return UC_EVENTBUS_QUEUE;
339 } 365 }
340 366
341 public String getEventBusSource(){ 367 public String getEventBusSource(){
342 if (MapUtils.isNotEmpty(eventBusInfo)) { 368 if (Objects.nonNull(eventBusInfo)) {
343 String source = eventBusInfo.get("active"); 369 if (MapUtils.isNotEmpty(eventBusInfo)) {
344 if (StringUtils.isNotBlank(source)) { 370 String source = eventBusInfo.get("active");
345 return this.chargeSource(source); 371 if (StringUtils.isNotBlank(source)) {
346 } else { 372 return this.chargeSource(source);
347 return ""; 373 } else {
374 return SERVICE_;
375 }
348 } 376 }
349 } 377 }
350 378
351 return null; 379 return SERVICE_;
352 } 380 }
353 381
354 public String getEventBusStartUp(){ 382 public String getEventBusStartUp(){
355 if (MapUtils.isNotEmpty(eventBusInfo)) { 383 if (Objects.nonNull(eventBusInfo)) {
356 String source = eventBusInfo.get("active"); 384 if (MapUtils.isNotEmpty(eventBusInfo)) {
357 if (StringUtils.isNotBlank(source)) { 385 String source = eventBusInfo.get("active");
358 return "true"; 386 if (StringUtils.isNotBlank(source)) {
387 return "true";
388 }
359 } 389 }
360 } 390 }
361
362 return "false"; 391 return "false";
363 } 392 }
364 393
...@@ -371,32 +400,37 @@ public class RabbitMqSourceConfig { ...@@ -371,32 +400,37 @@ public class RabbitMqSourceConfig {
371 private Map<String, String> wechatInfo; 400 private Map<String, String> wechatInfo;
372 401
373 public String getWechatQueue(){ 402 public String getWechatQueue(){
374 if (MapUtils.isNotEmpty(wechatInfo)) { 403 if (Objects.nonNull(wechatInfo)) {
375 String queue = wechatInfo.get("queue"); 404 if (MapUtils.isNotEmpty(wechatInfo)) {
376 return queue; 405 String queue = wechatInfo.get("queue");
406 return queue;
407 }
377 } 408 }
378
379 return WEIXIN_SUBORUNSUB_QUEUE; 409 return WEIXIN_SUBORUNSUB_QUEUE;
380 } 410 }
381 411
382 public String getWechatSource(){ 412 public String getWechatSource(){
383 if (MapUtils.isNotEmpty(wechatInfo)) { 413 if (Objects.nonNull(wechatInfo)) {
384 String source = wechatInfo.get("active"); 414 if (MapUtils.isNotEmpty(wechatInfo)) {
385 if (StringUtils.isNotBlank(source)) { 415 String source = wechatInfo.get("active");
386 return this.chargeSource(source); 416 if (StringUtils.isNotBlank(source)) {
387 } else { 417 return this.chargeSource(source);
388 return ""; 418 } else {
419 return SERVICE_;
420 }
389 } 421 }
390 } 422 }
391 423
392 return null; 424 return SERVICE_;
393 } 425 }
394 426
395 public String getWechatStartUp(){ 427 public String getWechatStartUp(){
396 if (MapUtils.isNotEmpty(wechatInfo)) { 428 if (Objects.nonNull(wechatInfo)) {
397 String source = wechatInfo.get("active"); 429 if (MapUtils.isNotEmpty(wechatInfo)) {
398 if (StringUtils.isNotBlank(source)) { 430 String source = wechatInfo.get("active");
399 return "true"; 431 if (StringUtils.isNotBlank(source)) {
432 return "true";
433 }
400 } 434 }
401 } 435 }
402 436
......
...@@ -79,7 +79,7 @@ mutil-mq: ...@@ -79,7 +79,7 @@ mutil-mq:
79 # password: Topdraw1qaz 79 # password: Topdraw1qaz
80 # 虚拟空间 80 # 虚拟空间
81 # virtual-host: member_center_iptv_sichuan 81 # virtual-host: member_center_iptv_sichuan
82 virtual-host: member_center_iptv_chongshu 82 virtual-host: user_center
83 publisher-confirms: true #如果对异步消息需要回调必须设置为true 83 publisher-confirms: true #如果对异步消息需要回调必须设置为true
84 84
85 # 管理侧 85 # 管理侧
...@@ -99,42 +99,60 @@ mutil-mq: ...@@ -99,42 +99,60 @@ mutil-mq:
99 service: 99 service:
100 mq: 100 mq:
101 list: 101 list:
102 - source: event 102 # - source: event
103 exchange: event.exchange 103 # exchange: event.exchange
104 queue: event.queue 104 # queue: event.queue
105 # exchange-type: direct
106 # routing-key:
107 # active: service
108 - source: collection
109 exchange: userCenter_exchange
110 queue: queue.collection.add
105 exchange-type: direct 111 exchange-type: direct
106 routing-key: 112 routing-key: route.UserCollection.add
107 active: service 113 active: service
108 - source: collection 114 - source: collection
109 exchange: collection.exchange 115 exchange: userCenter_exchange
110 queue: collection.queue 116 queue: queue.collection.delete
111 exchange-type: direct 117 exchange-type: direct
112 routing-key: service 118 routing-key: route.UserCollection.delete
113 active: service 119 active: service
114 - source: viewRecord 120 - source: collection
115 exchange: viewRecord.exchange 121 exchange: userCenter_exchange
116 queue: viewRecord.queue 122 queue: queue.collection.deleteall
117 exchange-type: direct 123 exchange-type: direct
118 routing-key: 124 routing-key: route.UserCollection.deleteall
119 active: service 125 active: service
120 - source: eventBus 126 - source: collection
121 exchange: uc.eventbus 127 exchange: exchange.collection
122 queue: uc.eventbus 128 queue: collection.queue
123 exchange-type: topic 129 exchange-type: direct
124 routing-key: uc.eventbus.*.topic 130 routing-key:
125 active: service 131 active: service
132 # - source: viewRecord
133 # exchange: viewRecord.exchange
134 # queue: viewRecord.queue
135 # exchange-type: direct
136 # routing-key:
137 # active: service
138 # - source: eventBus
139 # exchange: uc.eventbus
140 # queue: uc.eventbus
141 # exchange-type: topic
142 # routing-key: uc.eventbus.*.topic
143 # active: service
126 - source: uce 144 - source: uce
127 exchange: uce.exchange 145 exchange: uce.exchange
128 queue: uce.queue 146 queue: uce.queue
129 exchange-type: direct 147 exchange-type: direct
130 routing-key: 148 routing-key:
131 active: management 149 active: management
132 - source: wechat 150 # - source: wechat
133 exchange: weixin.subOrUnSub.direct 151 # exchange: weixin.subOrUnSub.direct
134 queue: weixin.subOrUnSub.queue 152 # queue: weixin.subOrUnSub.queue
135 exchange-type: direct 153 # exchange-type: direct
136 routing-key: 154 # routing-key:
137 active: 155 # active: active
138 error: 156 error:
139 logs: 157 logs:
140 list: 158 list:
......