[JAVA] Essayez d'utiliser MT Loader de Reladomo (chargeur de matrices multi-threads)
introduction
Dans la continuité de l'article que j'ai écrit plus tôt, cette fois j'ai essayé d'utiliser MT Loader de Reladomo.
Qu'est-ce que MT Loader (Multi-Threaded Matcher Loader)?
Le chargeur de matrices multi-threads (MT Loader) est une fonction permettant de fusionner les modifications d'une autre source (fichier, flux, autre DB, etc.) vers le DB de sortie. Après lecture de la documentation, MT Loader est souvent recommandé lorsque la quantité de données gérées par Reladomo est importante.
Image de fonctionnement de MT Loader
--Détecte les données qui existent à la fois dans les ensembles de données d'entrée et de sortie (base de données)
--Toutes les données qui existent dans les deux seront ʻUPDATE (cependant, s'il y a un changement dans les données) Ensembles de données ʻINSERT qui sont en entrée mais pas en sortie (base de données)
--Fermez tout ce qui est en sortie (base de données) mais pas en entrée (selon l'implémentation, DELETE ou expire)
Architecture de MT Loader
L'E / S peut être distribuée à l'aide de plusieurs threads pour la lecture, la comparaison et l'écriture.
Caractéristiques de MT Loader
--Il n'est pas nécessaire qu'il s'agisse d'un fichier à base de données. Il peut être facilement changé en Database-to-Database ou Memory-to-Database.
- En sous-classant MatcherThread ou SingleQueueExecutor, il est possible de régler selon diverses exigences.
- Écrit uniquement dans une table. Il est possible d'écrire dans plusieurs tables, mais il n'y a aucune garantie de transactions. C'est un atout dans les cas d'utilisation qui traitent de grandes quantités de données, il ne peut donc pas être utilisé dans les cas où la gestion des transactions est importante. (Soyez prudent lorsque vous l'utilisez)
- La conception peut garantir l'égalité.
Essayez d'utiliser MT Loader
J'écris souvent des API en combinaison avec Spring Boot, donc j'écrirai dans cette combinaison. Le code réel peut être trouvé à https://github.com/amtkxa/spring-boot-reladomo-mt-loader.
Écrire le code de test
Préparation préalable
Avant d'exécuter le code de test, je voudrais créer un état dans lequel les données de test ont été lues à l'avance dans le DB de test. Puisqu'il est supposé que MithraTestResource.addTestDataToDatabase sera utilisé pour lire les données de test, préparez le fichier suivant.
customer_data.txt
class com.amtkxa.domain.entity.Customer
customerId, name, country, businessDateFrom, businessDateTo, processingDateFrom, processingDateTo
1,"Liam","USA","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000"
2,"Emma","USA","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000"
3,"Noah","USA","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000"
4,"Olivia","USA","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000"
5,"William","USA","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000"
6,"Ava","USA","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000"
7,"James","USA","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000","2019-12-01 00:00:00.000","9999-12-01 23:59:00.000"
Nous avons également préparé une classe abstraite pour pré-remplir la base de données H2 avec des données de test.
public abstract class AbstractReladomoTest {
private static Logger log = LoggerFactory.getLogger(AbstractReladomoTest.class);
private MithraTestResource mithraTestResource;
protected abstract String[] getTestDataFilenames();
protected String getMithraConfigXmlFilename() {
return "reladomo/config/TestReladomoRuntimeConfiguration.xml";
}
@BeforeEach
public void setUp() {
log.info("Setting up reladomo on h2");
this.mithraTestResource = new MithraTestResource(this.getMithraConfigXmlFilename());
ConnectionManagerForTests connectionManager = ConnectionManagerForTests.getInstanceForDbName("testdb");
this.mithraTestResource.createSingleDatabase(connectionManager);
for (String filename : this.getTestDataFilenames()) {
this.mithraTestResource.addTestDataToDatabase(filename, connectionManager);
}
this.mithraTestResource.setUp();
}
@AfterEach
public void tearDown() {
this.mithraTestResource.tearDown();
}
}
Exécutez MT Loader avec le code de test
Donnez à MT Loader les données d'entrée suivantes et essayez de mettre à jour le DB.
--customerId: pays mis à jour pour 6 utilisateurs --Ajouter un nouvel utilisateur
- D'autres utilisateurs n'existent pas dans les données d'entrée (en attendant que la date d'expiration de l'utilisateur correspondant soit mise à jour pour expirer)
Le code de test que j'ai écrit est le suivant.
public class SingleQueueExecutorParallelLoadTest extends AbstractReladomoTest {
private static int NUMBER_OF_THREADS = 2;
private static int BATCH_SIZE = 5;
private static int INSERT_THREADS = 3;
@Override
public String[] getTestDataFilenames() {
return new String[] { "testdata/customer_data.txt" };
}
private List<Customer> getInputData() {
Timestamp businessDate = DateUtil.parse("2019-12-05 00:00:00");
CustomerList customerList = new CustomerList();
customerList.add(new Customer(6, "Ava", "JPN", businessDate));
customerList.add(new Customer(8, "Arthur", "USA", businessDate));
return customerList;
}
private CustomerList getDbRecords() {
return CustomerFinder.findMany(
CustomerFinder.all()
.and(CustomerFinder.businessDate().equalsEdgePoint())
.and(CustomerFinder.processingDate().equalsInfinity())
);
}
@Test
public void testLoadDataParallel() {
try {
QueueExecutor queueExecutor = new SingleQueueExecutor(
NUMBER_OF_THREADS,
CustomerFinder.customerId().ascendingOrderBy(),
BATCH_SIZE,
CustomerFinder.getFinderInstance(),
INSERT_THREADS
);
MatcherThread<Customer> matcherThread = new MatcherThread<>(
queueExecutor,
new Extractor[] { CustomerFinder.customerId() }
);
matcherThread.start();
// Database data load: Parallel
DbLoadThread dbLoadThread = new DbLoadThread(getDbRecords(), null, matcherThread);
dbLoadThread.start();
// Input data load: Parallel
PlainInputThread inputThread = new PlainInputThread(new InputDataLoader(), matcherThread);
inputThread.run();
matcherThread.waitTillDone();
// Assert
checkResult(queueExecutor);
} catch (Exception e) {
throw new ReladomoMTLoaderException("Failed to load data. " + e.getMessage(), e.getCause());
}
}
private void checkResult(QueueExecutor queueExecutor) {
// Whatever is in Output Set but not in Input Set will be closed out (terminated).
CustomerList customerList = getDbRecords();
assertEquals(2, customerList.count());
// Whatever is in the intersection, will be updated (but only if something changed)
Customer customer = CustomerFinder.findOne(
CustomerFinder.customerId().eq(6)
.and(CustomerFinder.businessDate().equalsEdgePoint())
.and(CustomerFinder.processingDate().equalsInfinity())
);
assertAll("Check updated customer data",
() -> assertEquals("Ava", customer.getName()),
() -> assertEquals("JPN", customer.getCountry()) // Updated from USD to JPN
);
// Whatever in in Input Set but not in Output Set will be inserted
Customer customer8 = CustomerFinder.findOne(
CustomerFinder.customerId().eq(8)
.and(CustomerFinder.businessDate().equalsEdgePoint())
.and(CustomerFinder.processingDate().equalsInfinity())
);
assertAll("Check inserted customer data",
() -> assertEquals("Arthur", customer8.getName()),
() -> assertEquals("USA", customer8.getCountry())
);
assertAll("Check the count of inserts, updates, terminates",
() -> assertEquals(1, queueExecutor.getTotalInserts()),
() -> assertEquals(1, queueExecutor.getTotalUpdates()),
() -> assertEquals(6, queueExecutor.getTotalTerminates())
);
}
private class InputDataLoader implements InputLoader {
private boolean firstTime = true;
@Override
public List<? extends MithraTransactionalObject> getNextParsedObjectList() {
return getInputData();
}
@Override
public boolean isFileParsingComplete() {
if (firstTime) {
firstTime = false;
return false;
} else {
return true;
}
}
}
}
Avant que le processus de mise à jour par MT Loader ne fonctionne, le DB se trouve dans l'état suivant.
-- select * from customer where business_date_to = '9999-12-01 23:59:00.000' and processing_date_to = '9999-12-01 23:59:00.000';
+-------------+---------+---------+-------------------------+-------------------------+-------------------------+-------------------------+
| customer_id | name | country | business_date_from | business_date_to | processing_date_from | processing_date_to |
+-------------+---------+---------+-------------------------+-------------------------+-------------------------+-------------------------+
| 1 | Liam | USA | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 |
| 2 | Emma | USA | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 |
| 3 | Noah | USA | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 |
| 4 | Olivia | USA | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 |
| 5 | William | USA | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 |
| 6 | Ava | USA | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 |
| 7 | James | USA | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 | 2019-12-01 00:00:00.000 | 9999-12-01 23:59:00.000 |
+-------------+---------+---------+-------------------------+-------------------------+-------------------------+-------------------------+
Une fois que le processus de mise à jour par MT Loader a fonctionné, la base de données était dans l'état suivant et le résultat était comme prévu.
-- select * from customer where business_date_to = '9999-12-01 23:59:00.000' and processing_date_to = '9999-12-01 23:59:00.000';
+-------------+--------+---------+-------------------------+-------------------------+-------------------------+-------------------------+
| customer_id | name | country | business_date_from | business_date_to | processing_date_from | processing_date_to |
+-------------+--------+---------+-------------------------+-------------------------+-------------------------+-------------------------+
| 6 | Ava | JPN | 2020-10-10 14:24:52.387 | 9999-12-01 23:59:00.000 | 2020-10-10 14:24:53.250 | 9999-12-01 23:59:00.000 |
| 8 | Arthur | USA | 2020-10-10 14:24:52.387 | 9999-12-01 23:59:00.000 | 2020-10-10 14:24:53.250 | 9999-12-01 23:59:00.000 |
+-------------+--------+---------+-------------------------+-------------------------+-------------------------+-------------------------+
à la fin
Reladomo est l'une de mes technologies préférées que j'utilise beaucoup au travail, mais j'ai l'impression que les obstacles pour atteindre un point où elle peut être comprise et utilisée sont un peu élevés.
Par exemple ..... Il n'y a pas d'exemple dans kata qui peut être incorporé dans Spring Boot, qui est relativement souvent utilisé, et si vous essayez de l'utiliser, vous devez réfléchir à la façon d'incorporer DBConnectionManager. J'ai honte de dire que j'ai beaucoup de mal à faire quelque chose qui fonctionne en premier lieu.
Reladomo en lui-même est une très bonne technologie, mais je n'ai pas trouvé beaucoup d'articles à ce sujet ... J'avais l'impression que c'était un peu de gaspillage, alors cette fois je l'ai vérifié moi-même. J'ai décidé de tout partager ensemble.
Référencé
- https://github.com/goldmansachs/reladomo-kata
- Reladomo Kata
- Frequently Asked Questions
- Introducing Reladomo - Enterprise Open Source Java ORM, Batteries Included! (Part 2)
- ReladomoAdvancedUseCases.pdf
Recommended Posts