UcEngineManagement2IptvConsumer.java
2.86 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
package com.topdraw.mq.consumer;
import com.rabbitmq.client.Channel;
import com.topdraw.mq.domain.TableOperationMsg;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.util.FileUtil;
import com.topdraw.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import java.io.IOException;
import java.time.LocalDate;
import java.util.Map;
@Component
@Slf4j
public class UcEngineManagement2IptvConsumer {
@Autowired
AutoRoute autoUser;
@Autowired
RestTemplateClient restTemplateClient;
@Value("#{rabbitMqErrorLogConfig.getUceError()}")
private Map<String, String> error;
/**
* 事件
* @param content
* @description 基础数据同步
* @author Hongyan Wang
* @date 2021/9/7 11:26 上午
*/
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqSourceConfig.getUceQueue()}",
containerFactory = "#{rabbitMqSourceConfig.getUceSource()}",
autoStartup = "#{rabbitMqSourceConfig.getUceStartUp()}",
ackMode = "MANUAL")
public void ucEventConsumer(Channel channel, Message message, String content) throws IOException {
log.info(" receive dataSync msg , content is : {} ", content);
try {
TableOperationMsg tableOperationMsg = this.parseContent(content);
autoUser.route(tableOperationMsg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}
e.printStackTrace();
}
log.info("ucEventConsumer ====>>>> end");
}
/**
* 数据解析
* @param content
* @return
*/
private TableOperationMsg parseContent(String content) {
TableOperationMsg tableOperationMsg = JSONUtil.parseMsg2Object(content,TableOperationMsg.class);
Assert.notNull(tableOperationMsg,"ERROR -->> operationConsumer -->> parseContent -->> 【dataSyncMsg】 not be null !!");
return tableOperationMsg;
}
}