java - DynamoDB 仅部分更新
问题描述
我想使用 dynamoDBMapper 库通过 java 更新我的 DynamoDB。我所做的是将消息(我想要执行的更新)推送到一个 SQS,并让我的 java 代码使用这些消息并更新我的 dynamoDB。我发现当我使用脚本短时间推送超过 150 条消息时,所有数据都可以被消费,但 DynamoDB 中只有部分记录被更新。像这样更新 DynamoDB 的代码:
@Service
public class PersistenceMessageProcessingServiceImpl implements PersistenceMessageProcessingService{
@Override
public void process(TextMessage textMessage){
String eventData = textMessage.getText();
updateEventStatus(eventData);
}
/*
each input is a caseDetail messages in Event Table
get data, parse data and update relative records partially in dynamodb.
finally check if still any open cases, if not change state of event
*/
private void updateEventStatus(String eventData) throws ParseException, IOException {
RetryUtils retryUtils = new RetryUtils(maxRetries, waitTimeInMilliSeconds, influxService);
SNowResponse serviceNowResponse = parseData(eventData);
EventCaseMap eventCaseMap = eventCaseMapRepository.findBySysId(sysId);
if (eventCaseMap != null) {
Event event = eventRepository.findByEventId(eventCaseMap.getSecurityEventManagerId());
CaseManagementDetails caseManagementDetails = event.getCaseManagementDetails();
Case existingCaseDetails = getCaseByCaseSystemId(caseManagementDetails, sysId);
caseDetails.setCaseStatus('Resolved');
caseDetails.setResolution(serviceNowResponse.getCloseCode());
caseDetails.setResolvedBy("A");
caseDetails.setAssessment(serviceNowResponse.getAssessment());
caseDetails.setResolutionSource("SEM");
retryUtils.run(() -> {
return eventRepository.updateEvent(event); }, RETRY_MEASUREMENT);
}
boolean stillOpen = false;
for(Case existingCase : caseManagementDetails.getCases()){
if(("OPEN").equals(existingCase.getCaseStatus().toString())){
stillOpen = true;
break;
}
}
if(!stillOpen){
event.setState('CLOSED');
}
}
private Case getCaseByCaseSystemId(CaseManagementDetails caseManagementDetails, String sysId) {
Case caseDetails = null;
if (caseManagementDetails != null) {
List<Case> caseList = caseManagementDetails.getCases();
for (Case c : caseList) {
if (c.getCaseSystemId() != null && c.getCaseSystemId().equalsIgnoreCase(sysId)) {
caseDetails = c;
break;
}
}
}
return caseDetails;
}
}
/* EventCaseMap Table in my DynamoDB
data model is like this for EventCaseMap Table:
{
"caseSystemId": "bb9cc488dbf67b40b3d57709af9619f8",
"securityEventManagerId": "756813a4-4e48-4abb-b37e-da00e931583b"
}
*/
@Repository
public class EventCaseMapRepositoryImpl implements EventCaseMapRepository {
@Autowired
DynamoDBMapper dynamoDBMapper;
@Override
public EventCaseMap findBySysId(String sysId) {
EventCaseMap eventCaseMap = new EventCaseMap();
eventCaseMap.setCaseSystemId(sysId);
return dynamoDBMapper.load(eventCaseMap, DynamoDBMapperConfig.ConsistentReads.CONSISTENT.config());
}
}
/*
data model is like this for Event Table:
{
"caseManagementDetails": {
"cases": [
{
"caseId": "SIR0123456",
"caseStatus": "OPEN",
},
{
"caseId": "SIR0654321",
"caseStatus": "OPEN",
},
{
many other cases(about two hundreds).....
}
]
},
"state": "OPEN",
"securityEventManagerId": "756813a4-4e48-4abb-b37e-da00e931583b"
}
*/
@Repository
public class EventRepositoryImpl implements EventRepository {
@Autowired
DynamoDBMapper dynamoDBMapper;
@Override
public Event findByEventId(String eventId) {
Event event = new Event();
event.setSecurityEventManagerId(eventId);
return dynamoDBMapper.load(event, DynamoDBMapperConfig.ConsistentReads.CONSISTENT.config());
}
@Override
public boolean updateEvent(Event event) {
dynamoDBMapper.save(event, DynamoDBMapperConfig.SaveBehavior.UPDATE_SKIP_NULL_ATTRIBUTES.config());
return false;
}
}
我已经尝试在 Intellij 的“RUN”和“DEBUG”模型中推送消息并一一使用消息。evertything 工作正常,所有案例都可以更新。所以我想知道 DynamoDB 中是否存在任何不一致问题,但我已经在我的代码中使用了强一致性。
那么有没有人知道我的代码中发生了什么?
有输入、输出、预期输出:
输入: 许多像这样的json文件:
{
"number": "SIR0123456",
"state": "Resolved",
"sys_id": "bb9cc488dbf67b40b3d57709af9619f8",
"MessageAttributes": {
"TransactionGuid": {
"Type": "String",
"Value": "093ddb36-626b-4ecc-8943-62e30ffa2e26"
}
}
}
{
"number": "SIR0654321",
"state": "Resolved",
"sys_id": "bb9cc488dbf67b40b3d57709af9619f7",
"MessageAttributes": {
"TransactionGuid": {
"Type": "String",
"Value": "093ddb36-626b-4ecc-8943-62e30ffa2e26"
}
}
}
事件表的输出:
{
"caseManagementDetails": {
"cases": [
{
"caseId": "SIR0123456",
"caseStatus": "RESOLVED",
},
{
"caseId": "SIR0654321",
"caseStatus": "OPEN"
},
{
many other cases(about two hundreds).....
}
]
},
"state": "OPEN",
"securityEventManagerId": "756813a4-4e48-4abb-b37e-da00e931583b"
}
事件表的预期输出:
{
"caseManagementDetails": {
"cases": [
{
"caseId": "SIR0123456",
"caseStatus": "RESOLVED",
},
{
"caseId": "SIR0654321",
"caseStatus": "RESOLVED"
},
{
many other cases(about two hundreds).....
}
]
},
"state": "OPEN",
"securityEventManagerId": "756813a4-4e48-4abb-b37e-da00e931583b"
}
解决方案
我认为问题在于当 DynamoDB 持久化数据时,它以多线程方式运行。所以如果我们在短时间内消耗所有这些数据,可能会有一些线程没有完成。所以我们看到的结果只是最后一个线程的结果,而不是所有线程的结果。
推荐阅读
- c++ - 如何从 Lua 访问运行 Lua 脚本的类的变量
- html - 来自磁盘缓存的 html5 applicationCache
- javascript - 我怎样才能获得特定收音机的价值
- indexing - WordDelimiterGraphFilterFactory 的 Solr 短语查询问题
- rxjs - 在 Redux Observable 中,如何在 img 加载后获取图像大小
- dart - 在 Dart 中检测动态类型
- arrays - 将一维数组重塑为网格格式
- typo3 - TYPO3 9.2.1 在 Composer 模式下创建 ViewHelper
- angular - 在 typescript [ionic angular] 中执行 resolveLocalFilesystemUrl 后无法获取文件数据
- android - 如何在gradle中声明一个变量?