ApacheJena中Txn.executeWrite方法详解与代码示例

本文深入探讨了 Java 中 org.apache.jena.system.Txn.executeWrite() 方法的使用及其具体实现。该方法用于在写事务中执行指定的操作，确保数据的一致性和完整性。以下将通过一系列来自 Apache Jena 项目中的代码片段，展示 Txn.executeWrite() 的实际应用场景。

Txn.executeWrite 方法介绍

[英] Execute the Runnable in a write transaction
[中] 在写事务中执行 Runnable 操作

代码示例

示例 1：复制数据集

public static void copy(DatasetGraph dsgSrc, DatasetGraph dsgDst) {
  Txn.executeRead(dsgSrc, ()->{
    Txn.executeWrite(dsgDst, () -> {
      Iterator iter = dsgSrc.find();
      iter.forEachRemaining(dsgDst::add);
      copyPrefixes(dsgSrc, dsgDst);
    });
  });
}

示例 2：加载数据

private void load(String data) { 
  Txn.executeWrite(dataset, ()->{
    RDFDataMgr.read(dataset, new StringReader(data), null, Lang.TURTLE) ;
  }) ;
}

示例 3：加载数据集

@Override
public void loadDataset(Dataset dataset) {
  Txn.executeWrite(dataset,() ->{
    dataset.asDatasetGraph().find().forEachRemaining((q)->this.dataset.asDatasetGraph().add(q));
  });
}

示例 4：更新命名图

@Override
public void put(String graphName, Model model) {
  checkOpen();
  Txn.executeWrite(dataset, ()-> {
    Model modelDst = modelFor(graphName); 
    modelDst.removeAll();
    modelDst.add(model);
  });
}

示例 5：删除命名图

@Override
public void delete(String graph) {
  checkOpen();
  Txn.executeWrite(dataset,() ->{
    if ( LibRDFConn.isDefault(graph) ) 
      dataset.getDefaultModel().removeAll();
    else 
      dataset.removeNamedModel(graph);
  });
}

示例 6：添加四元组

private static void txnAddData(DatasetGraph dsg) {
    // Unique blank node.
    Quad q = SSE.parseQuad("(_ _:b :p 1)") ;
    Txn.executeWrite(dsg,  ()->dsg.add(q));
}

示例 7：测试事务管理器队列长度

@Test public void maxQueueThreshold_01() {
  TransactionManager.MaxQueueThreshold = 1 ;
  
  DatasetGraph dsg = create() ;
  TransactionManager tMgr = TDBInternal.getTransactionManager(dsg) ;
  Txn.executeWrite(dsg,  ()->{});
  assertEquals(1, tMgr.getQueueLength()) ;
  Txn.executeWrite(dsg,  ()->{});
  assertEquals(0, tMgr.getQueueLength()) ;
}

示例 8：处理异常

@Test(expected=ExceptionFromTest.class)
public void txn_exception_01() {
  Txn.executeWrite(counter, counter::inc) ;
  
  Txn.executeWrite(counter, () -> {
    counter.inc() ;
    assertEquals("In W, value()", 1, counter.value()) ;
    assertEquals("In W, get()",2, counter.get()) ;
    throw new ExceptionFromTest() ;
  }) ;
}

示例 9：批量操作

@Test public void queueBatchSize_01() {
  TransactionManager.QueueBatchSize = 0 ; // Immediate.
  
  DatasetGraph dsg = create() ;
  TransactionManager tMgr = TDBInternal.getTransactionManager(dsg) ;
  Txn.executeWrite(dsg,  ()->{});
  assertEquals(0, tMgr.getQueueLength()) ;
  Txn.executeWrite(dsg,  ()->{});
  assertEquals(0, tMgr.getQueueLength()) ;
}

示例 10：文件传输

@Test public void transObjectFile_1() {
  Txn.executeWrite(transactional, ()->{
    long x = transBinData.write(bytes1) ;
    assertEquals(0L, x) ;
    assertEquals(len1, transBinData.length()) ;
  }) ;
}

示例 11：嵌套事务

@Test
public void txn_nested_13() {
  Txn.exec(counter, TxnType.READ_COMMITTED_PROMOTE, ()->{
    // Does not block promote
    ThreadAction.create(
      ()->Txn.executeWrite(counter, ()->counter.inc())
      ).run();
    // Can promote outer.
    Txn.exec(counter, TxnType.WRITE, ()->{});
  });
}

示例 12：查询和构造

@Test public void query_construct_01() {
  try ( RDFConnection cOnn= connection() ) {
    Txn.executeWrite(conn, ()->conn.loadDataset(DIR+"data.trig"));
    Txn.executeRead(conn, ()-> {
      Model m = conn.queryConstruct("CONSTRUCT WHERE { ?s ?p ?o }");
      assertEquals(2, m.size());
    });
  }
}

示例 13：基本事务操作

@Test public void txn_basic_03() {
  Txn.executeWrite(counter, counter::inc) ;
  long x = 
    Txn.calculateRead(counter, () -> {
      assertEquals("In R, value()", 1, counter.value()) ;
      assertEquals("In R, get()", 1, counter.get()) ;
      return counter.get() ;
    }) ;
  assertEquals("Outside R", 1, x) ;
}

示例 14：线程事务

@Test public void libTxnThread_12() {
    long x1 = counter1.get() ; 
    ThreadAction t = ThreadTxn.threadTxnRead(unit, () -> {
      long z1 = counter1.get() ;
      assertEquals("Thread", x1, z1) ;
    }) ;
    Txn.executeWrite(unit, ()->counter1.inc()) ;
    t.run() ;
    long x2 = counter1.get() ;
    assertEquals("after::", x1+1, x2) ;
}

示例 15：数据库管理

@Test
public void testDatabaseMgr1() {
  TDBInternal.reset() ; 
  DatasetGraph dg1 = DatabaseMgr.connectDatasetGraph(Location.mem("FOO")) ;
  DatasetGraph dg2 = DatabaseMgr.connectDatasetGraph(Location.mem("FOO")) ;
  Txn.executeWrite(dg1, ()->{
    dg1.add(quad1) ;
  }) ;
  Txn.executeRead(dg2, ()->{
    assertTrue(dg2.contains(quad1)) ;
  }) ;
}