做网站虚拟主机和云服务器吗,做招标应该关注什么网站,企业网站怎么做推广,网站设置搜索关键字数据流处理
#1.概述
Ignite提供了一个数据流API#xff0c;可用于将大量连续的数据流注入Ignite集群#xff0c;数据流API支持容错和线性扩展#xff0c;并为注入Ignite的数据提供了至少一次保证#xff0c;这意味着每个条目至少会被处理一次。
数据通过与缓存关联的数据…数据流处理
#1.概述
Ignite提供了一个数据流API可用于将大量连续的数据流注入Ignite集群数据流API支持容错和线性扩展并为注入Ignite的数据提供了至少一次保证这意味着每个条目至少会被处理一次。
数据通过与缓存关联的数据流处理器流式注入到缓存中。数据流处理器自动缓冲数据并将其分组成批次以提高性能并将其并行发送到多个节点。
数据流API提供以下功能
添加到数据流处理器的数据将在节点之间自动分区和分布可以以并置方式并发处理数据客户端可以在注入数据时对数据执行并发SQL查询。 #2.数据流处理器
数据流处理器与某个缓存关联并提供用于将数据注入缓存的接口。
在典型场景中用户拿到数据流处理器之后会使用其中某个方法将数据流式注入缓存中而Ignite根据分区规则对数据条目进行批处理从而避免不必要的数据移动。
拿到某个缓存的数据流处理器的方法如下
JavaC#/.NET
// Get the data streamer reference and stream data.
try (IgniteDataStreamerInteger, String stmr ignite.dataStreamer(myCache)) {// Stream entries.for (int i 0; i 100000; i)stmr.addData(i, Integer.toString(i));
}
System.out.println(dataStreamerExample output: cache.get(99999));在Ignite的Java版本中数据流处理器是IgniteDataStreamer接口的实现IgniteDataStreamer提供了一组addData(…)方法来向缓存中添加键-值对完整的方法列表可以参见IgniteDataStreamer的javadoc。
#3.覆写已有的数据
数据流处理器默认不会覆盖已有的数据通过将allowOverwrite属性配置为true可以修改该行为。
JavaC#/.NET
stmr.allowOverwrite(true);提示
如果allowOverwrite配置为false默认更新不会传播到外部存储如果开启。
#4.处理数据
如果需要在添加新数据之前执行自定义逻辑则可以使用数据流接收器。在将数据存储到缓存之前数据流接收器用于以并置方式处理数据其中实现的逻辑会在存储数据的节点上执行。
JavaC#/.NET
try (IgniteDataStreamerInteger, String stmr ignite.dataStreamer(myCache)) {stmr.allowOverwrite(true);stmr.receiver((StreamReceiverInteger, String) (cache, entries) - entries.forEach(entry - {// do something with the entrycache.put(entry.getKey(), entry.getValue());}));
}提示
注意数据流接收器不会自动将数据注入缓存需要显式地调用put(…)方法之一。
警告
要在远端节点执行的接收器类定义必须在该节点可用这可通过2种方式实现
将类文件加入该节点的类路径开启对等类加载
#4.1.StreamTransformer
StreamTransformer是StreamReceiver的简单实现用于更新流中的数据。数据流转换器利用了并置的特性并在将要存储数据的节点上更新数据。
在下面的示例中使用StreamTransformer为文本流中找到的每个不同单词增加一个计数
JavaC#/.NET
String[] text { hello, world, hello, Ignite };
CacheConfigurationString, Long cfg new CacheConfiguration(wordCountCache);IgniteCacheString, Long stmCache ignite.getOrCreateCache(cfg);try (IgniteDataStreamerString, Long stmr ignite.dataStreamer(stmCache.getName())) {// Allow data updates.stmr.allowOverwrite(true);// Configure data transformation to count instances of the same word.stmr.receiver(StreamTransformer.from((e, arg) - {// Get current count.Long val e.getValue();// Increment count by 1.e.setValue(val null ? 1L : val 1);return null;}));// Stream words into the streamer cache.for (String word : text)stmr.addData(word, 1L);}#4.2.StreamVisitor
StreamVisitor也是StreamReceiver的一个方便实现它会访问流中的每个键-值对但不会更新缓存。如果键-值对需要存储在缓存内那么需要显式地调用任意的put(...)方法。
在下面的示例中有两个缓存:marketData和instruments收到market数据的瞬间就会将它们放入marketData缓存的流处理器映射到某market数据的集群节点上的marketData的流处理器的StreamVisitor就会被调用在分别收到market数据后就会用最新的市场价格更新instrument缓存。
注意根本不会更新marketData缓存它一直是空的只是直接在数据将要存储的集群节点上简单利用了market数据的并置处理能力。
JavaC#/.NET
static class Instrument {final String symbol;Double latest;Double high;Double low;public Instrument(String symbol) {this.symbol symbol;}}static MapString, Double getMarketData() {//populate market data somehowreturn new HashMap();
}Test
void streamVisitorExample() {try (Ignite ignite Ignition.start()) {CacheConfigurationString, Double mrktDataCfg new CacheConfiguration(marketData);CacheConfigurationString, Instrument instCfg new CacheConfiguration(instruments);// Cache for market data ticks streamed into the system.IgniteCacheString, Double mrktData ignite.getOrCreateCache(mrktDataCfg);// Cache for financial instruments.IgniteCacheString, Instrument instCache ignite.getOrCreateCache(instCfg);try (IgniteDataStreamerString, Double mktStmr ignite.dataStreamer(marketData)) {// Note that we do not populate the marketData cache (it remains empty).// Instead we update the instruments cache based on the latest market price.mktStmr.receiver(StreamVisitor.from((cache, e) - {String symbol e.getKey();Double tick e.getValue();Instrument inst instCache.get(symbol);if (inst null)inst new Instrument(symbol);// Update instrument price based on the latest market tick.inst.high Math.max(inst.high, tick);inst.low Math.min(inst.low, tick);inst.latest tick;// Update the instrument cache.instCache.put(symbol, inst);}));// Stream market data into the cluster.MapString, Double marketData getMarketData();for (Map.EntryString, Double tick : marketData.entrySet())mktStmr.addData(tick);}}
}#5.配置数据流处理器线程池大小
数据流处理器线程池专用于处理来自数据流处理器的消息。
默认池大小为max(8, CPU总核数)使用IgniteConfiguration.setDataStreamerThreadPoolSize(…)可以改变线程池的大小。
XMLJava
IgniteConfiguration cfg new IgniteConfiguration();
cfg.setDataStreamerThreadPoolSize(10);Ignite ignite Ignition.start(cfg);键-值API
#1.基本缓存操作
#1.1.获取缓存的实例
在缓存上的所有操作都是通过IgniteCache实例进行的也可以在已有的缓存上拿到IgniteCache也可以动态创建。
JavaC#/.NETC
Ignite ignite Ignition.ignite();// Obtain an instance of the cache named myCache.
// Note that different caches may have different generics.
IgniteCacheInteger, String cache ignite.cache(myCache);#1.2.动态创建缓存
动态创建缓存方式如下
JavaC#/.NETC
Ignite ignite Ignition.ignite();CacheConfigurationInteger, String cfg new CacheConfiguration();cfg.setName(myNewCache);
cfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);// Create a cache with the given name if it does not exist.
IgniteCacheInteger, String cache ignite.getOrCreateCache(cfg);关于缓存的配置参数请参见缓存配置章节的内容。
在基线拓扑变更过程中调用创建缓存的方法会抛出org.apache.ignite.IgniteCheckedException异常
javax.cache.CacheException: class org.apache.ignite.IgniteCheckedException: Failed to start/stop cache, cluster state change is in progress.at org.apache.ignite.internal.processors.cache.GridCacheUtils.convertToCacheException(GridCacheUtils.java:1323)at org.apache.ignite.internal.IgniteKernal.createCache(IgniteKernal.java:3001)at org.apache.ignite.internal.processors.platform.client.cache.ClientCacheCreateWithNameRequest.process(ClientCacheCreateWithNameRequest.java:48)at org.apache.ignite.internal.processors.platform.client.ClientRequestHandler.handle(ClientRequestHandler.java:51)at org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.onMessage(ClientListenerNioListener.java:173)at org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.onMessage(ClientListenerNioListener.java:47)at org.apache.ignite.internal.util.nio.GridNioFilterChain$TailFilter.onMessageReceived(GridNioFilterChain.java:278)at org.apache.ignite.internal.util.nio.GridNioFilterAdapter.proceedMessageReceived(GridNioFilterAdapter.java:108)at org.apache.ignite.internal.util.nio.GridNioAsyncNotifyFilter$3.body(GridNioAsyncNotifyFilter.java:96)at org.apache.ignite.internal.util.worker.GridWorker.run(GridWorker.java:119)at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)at java.base/java.lang.Thread.run(Thread.java:834)如果拿到这个异常可以进行重试。
#1.3.销毁缓存
要在整个集群中删除一个缓存需要调用destroy()方法
Java
Ignite ignite Ignition.ignite();IgniteCacheLong, String cache ignite.cache(myCache);cache.destroy();#1.4.原子化操作
拿到缓存实例后就可以对其进行读写操作
JavaC#/.NETC
IgniteCacheInteger, String cache ignite.cache(myCache);// Store keys in the cache (the values will end up on different cache nodes).
for (int i 0; i 10; i)cache.put(i, Integer.toString(i));for (int i 0; i 10; i)System.out.println(Got [key i , val cache.get(i) ]);提示
putAll()和putAll()这样的批量操作方法是以原子化的模式按顺序执行可能部分失败。发生这种情况时会抛出包含了更新失败数据列表的CachePartialUpdateException异常。 如果希望在一个操作中更新条目的集合建议考虑使用事务。
下面是更多基本缓存操作的示例
JavaC#/.NETC
// Put-if-absent which returns previous value.
String oldVal cache.getAndPutIfAbsent(11, Hello);// Put-if-absent which returns boolean success flag.
boolean success cache.putIfAbsent(22, World);// Replace-if-exists operation (opposite of getAndPutIfAbsent), returns previous
// value.
oldVal cache.getAndReplace(11, New value);// Replace-if-exists operation (opposite of putIfAbsent), returns boolean
// success flag.
success cache.replace(22, Other new value);// Replace-if-matches operation.
success cache.replace(22, Other new value, Yet-another-new-value);// Remove-if-matches operation.
success cache.remove(11, Hello);#1.5.异步执行
大多数缓存操作方法都有对应的异步执行模式方法名带有Async后缀。
JavaC#/.NETC
// a synchronous get
V get(K key);// an asynchronous get
IgniteFutureV getAsync(K key);异步操作会返回一个代表操作结果的对象可以以阻塞或非阻塞的方式等待操作的完成。
以非阻塞的方式等待结果可以使用IgniteFuture.listen()或IgniteFuture.chain()方法注册一个闭包其会在操作完成后被调用。
JavaC#/.NETC
IgniteCompute compute ignite.compute();// Execute a closure asynchronously.
IgniteFutureString fut compute.callAsync(() - Hello World);// Listen for completion and print out the result.
fut.listen(f - System.out.println(Job result: f.get()));闭包执行和线程池
如果在将闭包传递给IgniteFuture.listen()或IgniteFuture.chain()方法时已完成异步操作则该闭包由调用线程同步执行。否则当操作完成时闭包将异步执行。
根据操作的类型闭包将被系统线程池中的线程异步缓存操作或公共线程池中的线程异步计算操作调用。因此应避免在闭包内部调用同步缓存和计算操作因为由于线程池不足它可能导致死锁。
为了实现异步计算操作的嵌套执行可以利用自定义线程池。
#2.使用二进制对象
#2.1.概述
在Ignite中数据以二进制格式存储然后在每次读取时再反序列化为对象不过可以直接操作二进制对象避免反序列化。
二进制对象是缓存数据的二进制表示的包装器每个二进制对象都有field(name)方法返回对应字段的值和type()方法提取对象的类型信息。当只需要处理对象的部分字段而不需要反序列化整个对象时二进制对象会很有用。
处理二进制对象时不需要具体的类定义不重启集群就可以动态修改对象的结构。
在所有支持的平台上二进制对象格式都是统一的包括Java、.NET和C。可以启动一个Java版Ignite集群然后使用.NET和C客户端接入集群然后在这些客户端上使用二进制对象而不需要持有类定义。
限制
在内部二进制对象的类型和字段以ID来标识该ID由对应字符串名字的哈希值计算得出这意味着属性或者类型不能有同样的名字哈希因此不允许使用具有相同名字哈希的字段或类型。但是可以通过配置提供自定义的ID生成实现同样的原因BinaryObject格式在类的不同层次上也不允许有同样的属性名如果类实现了Externalizable接口Ignite会使用OptimizedMarshallerOptimizedMarshaller会使用writeExternal()和readExternal()来进行类对象的序列化和反序列化这需要将实现Externalizable的类加入服务端节点的类路径中。
#2.2.启用缓存的二进制模式
当从缓存中拿数据时默认返回的是反序列化格式要处理二进制格式需要使用withKeepBinary()方法拿到缓存的实例这个实例会尽可能返回二进制格式的对象。
JavaC#/.NET
// Create a regular Person object and put it into the cache.
Person person new Person(1, FirstPerson);
ignite.cache(personCache).put(1, person);// Get an instance of binary-enabled cache.
IgniteCacheInteger, BinaryObject binaryCache ignite.cache(personCache).withKeepBinary();
BinaryObject binaryPerson binaryCache.get(1);注意并不是所有的对象都会转为二进制对象格式下面的类不会进行转换即toBinary(Object)方法返回原始对象以及这些类的实例存储不会发生变化
所有的基本类型byte、int等及其包装类Byte、Integer等基本类型的数组byte[]、int[]等String及其数组UUID及其数组Date及其数组Timestamp及其数组Enum及其数组对象的映射、数组和集合但如果它们是可以转成二进制的则内部对象将被重新转换。
#2.3.创建和修改二进制对象
二进制对象实例是不可变的要更新字段或者创建新的二进制对象需要使用二进制对象的建造器工具类其可以在没有对象的类定义的前提下修改二进制对象的字段。
限制
无法修改已有字段的类型无法变更枚举值的顺序也无法在枚举值列表的开始或者中部添加新的常量但是可以在列表的末尾添加新的常量。
二进制对象建造器实例获取方式如下
JavaC#/.NET
BinaryObjectBuilder builder ignite.binary().builder(org.apache.ignite.snippets.Person);builder.setField(id, 2L);
builder.setField(name, SecondPerson);binaryCache.put(2, builder.build());通过这个方式创建的建造器没有任何字段调用setField(…)方法可以添加字段
通过调用toBuilder()方法也可以从一个已有的二进制对象上获得建造器实例这时该二进制对象的所有字段都会复制到该建造器中。
在下面的示例中会在服务端通过EntryProcessor机制更新一个对象而不需要在该节点部署该对象类定义也不需要完整对象的反序列化。
Java
// The EntryProcessor is to be executed for this key.
int key 1;
ignite.cache(personCache).Integer, BinaryObjectwithKeepBinary().invoke(key, (entry, arguments) - {// Create a builder from the old value.BinaryObjectBuilder bldr entry.getValue().toBuilder();//Update the field in the builder.bldr.setField(name, Ignite);// Set new value to the entry.entry.setValue(bldr.build());return null;
});#2.4.二进制类型和二进制字段
二进制对象持有其表示的对象的类型信息类型信息包括字段名、字段类型和关联字段名。
每个字段的类型通过一个BinaryField对象来表示获得BinaryField对象后如果需要从集合中的每个对象读取相同的字段则可以多次重用该对象。重用BinaryField对象比直接从每个二进制对象读取字段值要快下面是使用二进制字段的示例
CollectionBinaryObject persons getPersons();BinaryField salary null;
double total 0;
int count 0;for (BinaryObject person : persons) {if (salary null) {salary person.type().field(salary);}total (float) salary.value(person);count;
}double avg total / count;#2.5.二进制对象的调整建议
Ignite为给定类型的每个二进制对象保留一个模式该模式指定对象中的字段及其顺序和类型。模式在整个集群中复制具有相同字段但顺序不同的二进制对象被认为具有不同的模式因此建议以相同的顺序往二进制对象中添加字段。
空字段通常需要5个字节来存储字段ID4个字节字段长度1个字节。在内存方面最好不要包含字段也不要包含空字段。但是如果不包括字段则Ignite会为此对象创建一个新模式该模式与包含该字段的对象的模式不同。如果有多个字段以随机组合设置为null那么Ignite会为每种组合维护一个不同的二进制对象模式这样Java堆可能会被二进制对象模式耗尽。最好为二进制对象提供几个模式并以相同的顺序设置相同类型的相同字段集。通过提供相同的字段集即使具有空值来创建二进制对象时选择其中一个这也是需要为空字段提供字段类型的原因。
如果有一个子集的字段是可选的但要么全部不存在要么全部存在那么也可以嵌套二进制对象可以将它们放在单独的二进制对象中该对象存储在父对象的字段下或者设置为null。
如果有大量字段这些字段在任何组合中都是可选的并且通常为空则可以将其存储在映射字段中值对象中将有几个固定字段还有一个映射用于其他属性。
#2.6.配置二进制对象
在绝大多数场景中无需配置二进制对象。但是如果需要更改类型和字段ID的生成或插入自定义序列化器则可以通过配置来实现。
二进制对象的类型和字段由其ID标识该ID由相对应的字符串名计算为哈希值并将其存储在每个二进制对象中可以在配置中定义自己的ID生成实现。
名字到ID的转换分为两个步骤。首先由名字映射器转换类型名类名或字段名然后由ID映射器计算ID。可以指定全局名字映射器全局ID映射器和全局二进制序列化器以及每个类型的映射器和序列化器。每个类型的配置均支持通配符这时所提供的配置将应用于与类型名字模板匹配的所有类型。
XMLJavaC#/.NET
IgniteConfiguration igniteCfg new IgniteConfiguration();BinaryConfiguration binaryConf new BinaryConfiguration();
binaryConf.setNameMapper(new MyBinaryNameMapper());
binaryConf.setIdMapper(new MyBinaryIdMapper());BinaryTypeConfiguration binaryTypeCfg new BinaryTypeConfiguration();
binaryTypeCfg.setTypeName(org.apache.ignite.snippets.*);
binaryTypeCfg.setSerializer(new ExampleSerializer());binaryConf.setTypeConfigurations(Collections.singleton(binaryTypeCfg));igniteCfg.setBinaryConfiguration(binaryConf);#3.使用扫描查询
#3.1.概述
IgniteCache有几个查询方法他们会接收Query类的子类然后返回一个QueryCursor。
Query表示在缓存上执行的分页查询的抽象页面大小通过Query.setPageSize(…)进行配置默认值为1024。
QueryCursor表示结果集可以透明地按页迭代。当用户迭代到页尾时QueryCursor会自动在后台请求下一页。对于不需要分页的场景可以使用QueryCursor.getAll()方法其会拿到所有的数据并将其存储在一个集合中。
关闭游标
调用QueryCursor.getAll()方法时游标会自动关闭。如果在循环中迭代游标或者显式拿到Iterator必须手动关闭游标或者使用try-with-resources语句。
#3.2.执行扫描查询
扫描查询是以分布式的方式从缓存中获取数据的简单搜索查询如果执行时没有参数扫描查询会从缓存中获取所有数据。
JavaC#/.NETC
IgniteCacheInteger, Person cache ignite.getOrCreateCache(myCache);QueryCursorCache.EntryInteger, Person cursor cache.query(new ScanQuery());如果指定了谓语扫描查询会返回匹配谓语的数据谓语应用于远端节点
JavaC#/.NET
IgniteCacheInteger, Person cache ignite.getOrCreateCache(myCache);// Find the persons who earn more than 1,000.
IgniteBiPredicateInteger, Person filter (key, p) - p.getSalary() 1000;try (QueryCursorCache.EntryInteger, Person qryCursor cache.query(new ScanQuery(filter))) {qryCursor.forEach(entry - System.out.println(Key entry.getKey() , Value entry.getValue()));
}扫描查询还支持可选的转换器闭包可以在数据返回之前在服务端转换数据比如当只想从大对象中获取少量字段以最小化网络传输时这个功能就很有用。下面的示例显示如何只返回键而不返回值
IgniteCacheInteger, Person cache ignite.getOrCreateCache(myCache);// Get only keys for persons earning more than 1,000.
ListInteger keys cache.query(new ScanQuery(// Remote filter(IgniteBiPredicateInteger, Person) (k, p) - p.getSalary() 1000),// Transformer(IgniteClosureCache.EntryInteger, Person, Integer) Cache.Entry::getKey).getAll();#3.3.本地扫描查询
扫描查询默认是分布到所有节点上的不过也可以只在本地执行查询这时查询只会处理本地节点查询执行的节点上存储的数据。
JavaC#/.NETC
QueryCursorCache.EntryInteger, Person cursor cache.query(new ScanQueryInteger, Person().setLocal(true));#3.4.相关主题
通过REST API执行扫描查询缓存查询事件。
#4.读修复
警告
这是个试验性API。
读修复是指在正常读取操作期间修复主备数据之间不一致的技术。当用户操作读取了某个或某些键时Ignite会检查给定键在所有备份副本中的值。
读修复模式旨在保持一致性。不过由于检查了备份副本因此读操作的成本增加了约2倍。通常不建议一直使用此模式而应一次性使用。
要启用读修复模式需要获取一个开启了读修复的缓存实例如下所示
IgniteCacheObject, Object cache ignite.cache(my_cache).withReadRepair();Object value cache.get(10);一致性检查与下面的缓存配置不兼容
没有备份的缓存本地缓存近缓存开启通读的缓存。
#4.1.事务化缓存
拓扑中的值将替换为最新版本的值。
对于配置为TransactionConcurrency.OPTIMISTIC并发模型或TransactionIsolation.READ_COMMITTED隔离级别的事务自动处理对于配置为TransactionConcurrency.PESSIMISTIC并发模型和TransactionIsolation.READ_COMMITTED隔离级别的事务在commit()阶段自动处理
当检测到备份不一致时Ignite将生成一个违反一致性事件如果在配置中启用了该事件通过监听该事件可以获取有关不一致问题的通知。关于如果进行事件监听请参见使用事件的介绍。
如果事务中已经缓存了值则读修复不能保证检查所有副本。例如如果使用非TransactionIsolation.READ_COMMITTED隔离级别并且已经读取了该值或执行了写入操作则将获得缓存的值。
#4.2.原子化缓存
如果发现差异则抛出违反一致性异常。
由于原子化缓存的性质可以观察到假阳性结果。比如在缓存加载中尝试检查一致性可能会触发违反一致性异常。读修复的实现会尝试检查给定键三次尝试次数可以通过IGNITE_NEAR_GET_MAX_REMAPS系统属性来修改。
注意不会为原子缓存记录违反一致性事件。 Ignite事务
#1.概述
要为某个缓存开启事务支持需要在缓存配置中将atomicityMode设置为TRANSACTIONAL具体请参见原子化模式。
事务可以将多个缓存操作可能对应一个或者多个键组合为一个单个原子事务这些操作在没有任何其他交叉操作的情况下执行或全部成功或全部失败没有部分成功的状态。
在缓存配置中可以为缓存开启事务支持
XMLJavaC#/.NET
CacheConfiguration cacheCfg new CacheConfiguration();cacheCfg.setName(cacheName);cacheCfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);IgniteConfiguration cfg new IgniteConfiguration();cfg.setCacheConfiguration(cacheCfg);// Optional transaction configuration. Configure TM lookup here.
TransactionConfiguration txCfg new TransactionConfiguration();cfg.setTransactionConfiguration(txCfg);// Start a node
Ignition.start(cfg);#2.执行事务
键-值API为开启和完成事务以及获取和事务有关的指标提供了一个接口该接口可以通过Ignite实例获得
JavaC#/.NETC
Ignite ignite Ignition.ignite();IgniteTransactions transactions ignite.transactions();try (Transaction tx transactions.txStart()) {Integer hello cache.get(Hello);if (hello 1)cache.put(Hello, 11);cache.put(World, 22);tx.commit();
}#3.并发模型和隔离级别
当原子化模式配置为TRANSACTIONAL时Ignite对事务支持乐观和悲观的并发模型。并发模型决定了何时获得一个条目级的事务锁-在访问数据时或者在prepare阶段。锁定可以防止对一个对象的并发访问。比如当试图用悲观锁更新一个ToDo列表项时服务端会在该对象上置一个锁在提交或者回滚该事务之前其它的事务或者操作都无法更新该条目。不管在一个事务中使用那种并发模型在提交之前都存在事务中的所有条目被锁定的时刻。
隔离级别定义了并发事务如何看以及处理针对同一个键的操作。Ignite支持READ_COMMITTED、REPEATABLE_READ、SERIALIZABLE隔离级别。
并发模型和隔离级别的所有组合都是可以同时使用的。下面是针对Ignite提供的每一个并发-隔离组合的行为和保证的描述。
#3.1.悲观事务
在PESSIMISTIC事务中锁是在第一次读或者写访问期间获得取决于隔离级别然后被事务持有直到其被提交或者回滚。该模式中锁首先在主节点获得然后在准备阶段提升至备份节点。下面的隔离级别可以配置为PESSIMISTIC并发模型。
READ_COMMITTED数据被无锁地读取并且不会被事务本身缓存。如果缓存配置允许数据是可能从一个备份节点中读取的。在这个隔离级别中可以有所谓的非可重复读因为当在自己的事务中读取数据两次时一个并发事务可以改变该数据。锁只有在第一次写访问时才会获得包括EntryProcessor调用。这意味着事务中已经读取的一个条目在该事务提交时可能有一个不同的值这种情况是不会抛出异常的。REPEATABLE_READ获得条目锁以及第一次读或者写访问时从主节点获得数据然后就存储在本地事务映射中。之后对同一数据的所有连续访问都是本地化的并且返回最后一次读或者被更新的事务值。这意味着没有其它的并发事务可以改变锁定的数据这样就获得了事务的可重复读。SERIALIZABLE在PESSIMISTIC模式中这个隔离级别与REPEATABLE_READ是一样的工作方式。
注意在PESSIMISTIC模式中锁的顺序是很重要的。此外Ignite可以按照指定的顺序依次并且准确地获得锁。
拓扑变化约束
注意如果至少获取了一个PESSIMISTIC事务锁则在提交或回滚事务之前将无法更改缓存拓扑。因此应该避免长时间持有事务锁。
#3.2.乐观事务
在OPTIMISTIC事务中条目锁是在二阶段提交的准备阶段从主节点获得的然后提升至备份节点该锁在事务提交时被释放。如果回滚事务没有试图做提交是不会获得锁的。下面的隔离级别可以与OPTIMISTIC并发模型配置在一起。
READ_COMMITTED应该作用于缓存的改变是在源节点上收集的然后事务提交后生效。事务数据无锁地读取并且不会在事务中缓存。如果缓存配置允许该数据是可能从备份节点中读取的。在这个隔离级别中可以有一个所谓的非可重复读因为在自己的事务中读取数据两次时另一个事务可以修改数据。这个模式组合在第一次读或者写操作后如果条目值被修改是不会做校验的并且不会抛出异常。REPEATABLE_READ这个隔离级别的事务的工作方式类似于OPTIMISTIC的READ_COMMITTED的事务只有一个不同读取值缓存于发起节点并且所有的后续读保证都是本地化的。这个模式组合在第一次读或者写操作后如果条目值被修改是不会做校验的并且不会抛出异常。SERIALIZABLE在第一次读访问之后会存储一个条目的版本如果Ignite引擎检测到发起事务中的条目只要有一个被修改Ignite就会在提交阶段放弃该事务这是在提交阶段对集群内的事务中记载的条目的版本进行内部检查实现的。简而言之这意味着Ignite如果在一个事务的提交阶段检测到一个冲突就会放弃这个事务并且抛出TransactionOptimisticException异常以及回滚已经做出的任何改变开发者应该处理这个异常并且重试该事务。
JavaC#/.NETC
CacheConfigurationInteger, String cfg new CacheConfiguration();
cfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
cfg.setName(myCache);
IgniteCacheInteger, String cache ignite.getOrCreateCache(cfg);// Re-try the transaction a limited number of times.
int retryCount 10;
int retries 0;// Start a transaction in the optimistic mode with the serializable isolation
// level.
while (retries retryCount) {retries;try (Transaction tx ignite.transactions().txStart(TransactionConcurrency.OPTIMISTIC,TransactionIsolation.SERIALIZABLE)) {// modify cache entries as part of this transaction.cache.put(1, foo);cache.put(2, bar);// commit the transactiontx.commit();// the transaction succeeded. Leave the while loop.break;} catch (TransactionOptimisticException e) {// Transaction has failed. Retry.}
}这里另外一个需要注意的重要的点是即使一个条目只是简单地读取没有改变cache.put(...)一个事务仍然可能失败因为该条目的值对于发起事务中的逻辑很重要。
注意对于READ_COMMITTED和REPEATABLE_READ事务键的顺序是很重要的因为这些模式中锁也是按顺序获得的。
#3.3.读一致性
为了在悲观模式下实现完全的读一致性需要获取读锁。这意味着只有在悲观的可重复读或可序列化事务中悲观模式下的读之间的完全一致性才能够实现。
当使用乐观事务时通过禁止读之间的潜在冲突可以实现完全的读一致性。此行为由乐观的可序列化模式提供。但是请注意在提交发生之前用户仍然可以读取部分事务状态因此事务逻辑必须对其进行保护。只有在提交阶段在任何冲突的情况下才会抛出TransactionOptimisticException这将允许开发者重试事务。
警告
如果没有使用悲观的可重复读或者可序列化事务或者也没有使用乐观的序列化事务那么就可以看到部分的事务状态这意味着如果一个事务更新对象A和B那么另一个事务可能看到A的新值和B的旧值。
#4.死锁检测
当处理分布式事务时必须要遵守的主要规则是参与一个事务的键的锁必须按照同样的顺序获得违反这个规则就可能导致分布式死锁。
Ignite无法避免分布式死锁而是有一个内建的功能来使调试和解决这个问题更容易。
在下面的代码片段中事务启动时带有超时限制如果到期死锁检测过程就会试图查找一个触发这个超时的可能的死锁。当超过超时时间时无论死锁如何都会向应用层抛出CacheException并将TransactionTimeoutException作为其触发原因。不过如果检测到了一个死锁返回的TransactionTimeoutException的触发原因会是TransactionDeadlockException至少涉及死锁的一个事务。
JavaC#/.NETC
CacheConfigurationInteger, String cfg new CacheConfiguration();
cfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
cfg.setName(myCache);
IgniteCacheInteger, String cache ignite.getOrCreateCache(cfg);try (Transaction tx ignite.transactions().txStart(TransactionConcurrency.PESSIMISTIC,TransactionIsolation.READ_COMMITTED, 300, 0)) {cache.put(1, 1);cache.put(2, 1);tx.commit();
} catch (CacheException e) {if (e.getCause() instanceof TransactionTimeoutException e.getCause().getCause() instanceof TransactionDeadlockException)System.out.println(e.getCause().getCause().getMessage());
}TransactionDeadlockException里面包含了有用的信息有助于找到导致死锁的原因。
Deadlock detected:K1: TX1 holds lock, TX2 waits lock.
K2: TX2 holds lock, TX1 waits lock.Transactions:TX1 [txIdGridCacheVersion [topVer74949328, time1463469328421, order1463469326211, nodeOrder1], nodeIdad68354d-07b8-4be5-85bb-f5f2362fbb88, threadId73]
TX2 [txIdGridCacheVersion [topVer74949328, time1463469328421, order1463469326210, nodeOrder1], nodeIdad68354d-07b8-4be5-85bb-f5f2362fbb88, threadId74]Keys:K1 [key1, cachedefault]
K2 [key2, cachedefault]死锁检测是一个多步过程根据集群中节点的数量、键以及可能导致死锁涉及的事务数可能需要做多次迭代。一个死锁检测的发起者是发起事务并且出现TransactionTimeoutException错误的那个节点这个节点会检查是否发生了死锁通过与其它远程节点交换请求/响应并且准备一个与死锁有关的、由TransactionDeadlockException提供的报告每个这样的消息请求/响应都会被称为一次迭代。
因为死锁检测过程不结束事务就不会回滚有时如果希望对于事务回滚有一个可预测的时间调整一下参数还是有意义的下面会描述。
IgniteSystemProperties.IGNITE_TX_DEADLOCK_DETECTION_MAX_ITERS指定死锁检测过程迭代的最大数如果这个属性的值小于等于0死锁检测会被禁用默认为1000IgniteSystemProperties.IGNITE_TX_DEADLOCK_DETECTION_TIMEOUT指定死锁检测机制的超时时间默认为1分钟。
注意如果迭代次数太少可能获得一个不完整的死锁检测报告。
#5.无死锁事务
对于OPTIMISTICSERIALIZABLE事务锁不是按顺序获得的。该模式中键可以按照任何顺序访问因为事务锁是通过一个额外的检查以并行的方式获得的这使得Ignite可以避免死锁。
这里需要引入几个概念来描述SERIALIZABLE的事务锁的工作方式。Ignite中的每个事务都会被赋予一个叫做XidVersion的可比较的版本号事务提交时该事务中修改的每个条目都会被赋予一个叫做EntryVersion的新的版本号一个版本号为XidVersionA的OPTIMISTICSERIALIZABLE事务在如下情况下会抛出TransactionOptimisticException异常而失败
有一个进行中的PESSIMISTIC或者非可序列化OPTIMISTIC事务在SERIALIZABLE事务中的一个条目上持有了一个锁有另外一个进行中的版本号为XidVersionB的OPTIMISTICSERIALIZABLE事务在XidVersionB XidVersionA时以及这个事务在SERIALIZABLE事务中的一个条目上持有了一个锁在该OPTIMISTICSERIALIZABLE事务获得所有必要的锁时存在在提交之前的版本与当前版本不同的条目
注意
在一个高并发环境中乐观锁可能出现高事务失败率而悲观锁如果锁被事务以一个不同的顺序获得可能导致死锁。
不过在一个同质化的环境中乐观可序列化锁对于大的事务可能提供更好的性能因为网络交互的数量只取决于事务相关的节点的数量而不取决于事务中的键的数量。
#6.处理失败事务
如下的异常可能导致事务失败
异常名称描述解决办法由TransactionTimeoutException触发的CacheException事务超时会触发TransactionTimeoutException。可以增加超时时间或者缩短事务执行时间TransactionDeadlockException触发TransactionTimeoutException再触发CacheException事务死锁会触发这个异常。使用死锁检测机制调试和修正死锁或者切换到乐观序列化事务无死锁事务。TransactionOptimisticException某种原因的乐观事务失败会抛出这个异常大多数情况下该异常发生在事务试图并发更新数据的场景中。重新执行事务。TransactionRollbackException事务自动或者手动回滚时可能抛出这个异常这时数据状态是一致的。因为数据状态是一致的所以可以对事务进行重试。TransactionHeuristicException这是一个不太可能发生的异常由Ignite中意想不到的内部错误或者通信错误导致该异常存在于事务子系统无法预知的不确定场景中目前没有被合理地处理。如果出现该异常数据可能不一致这时需要对数据进行重新加载或者报告给Ignite开发社区。
#7.长期运行事务终止
在Ignite集群中部分事件会触发分区映射的交换过程以及数据的再平衡来保证整个集群的数据分布这个事件的一个例子就是集群拓扑变更事件它会在新节点加入或者已有节点离开时触发还有新的缓存或者SQL表创建时也会触发分区映射的交换。
当分区映射交换开始时Ignite会在某个阶段拿到一个全局锁在未完成的事务并行执行时无法获得锁这些事务会阻止分区映射交换进程从而阻断一些新节点加入进程这样的一些操作。
使用TransactionConfiguration.setTxTimeoutOnPartitionMapExchange(...)方法可以配置长期运行事务阻断分区映射交换的最大时间时间一到所有的未完成事务都会回滚让分区映射交换进程先完成。
下面的示例显示如何配置超时时间
XMLJavaC#/.NET
// Create a configuration
IgniteConfiguration cfg new IgniteConfiguration();// Create a Transaction configuration
TransactionConfiguration txCfg new TransactionConfiguration();// Set the timeout to 20 seconds
txCfg.setTxTimeoutOnPartitionMapExchange(20000);cfg.setTransactionConfiguration(txCfg);// Start the node
Ignition.start(cfg);#8.事务监控
和事务相关的指标信息请参见事务监控章节的介绍。
关于如何追踪事务的信息请参见追踪章节的介绍。
另外还可以通过控制脚本获取事务的信息或者取消集群中正在执行的事务。
处理SQL
#1.介绍
Ignite是一个兼容于ANSI-99、可水平扩展和容错的分布式SQL数据库根据使用场景数据在整个集群中是以分区或者复制的模式进行分发。
作为SQL数据库Ignite支持所有DML命令包括SELECT、UPDATE、INSERT和DELETE语句并且还实现了与分布式系统相关的DDL命令的子集。
在外部工具和应用中通过使用JDBC或ODBC驱动可以像处理任何其他支持SQL的存储一样与Ignite交互。Java、.NET和C开发者还可以使用原生SQL API。
在内部SQL表具有与键-值缓存相同的数据结构这意味着可以更改数据的分区分布并利用关联并置技术获得更好的性能。
Ignite的SQL引擎使用H2数据库来解析和优化查询并生成执行计划。
#1.1.分布式查询
对分区表的查询以分布式方式执行
对该查询进行解析并分为多个映射查询和一个汇总查询所有映射查询都在数据所在的所有节点上执行所有节点都向查询发起方提供本地执行的结果集查询发起方会将各个结果集汇总为最终结果。
也可以强制查询在本地进行处理即在执行查询的节点上的数据子集上执行。
#1.2.本地查询
如果在复制表上执行查询将会在本地数据上执行。
#2.理解模式
#2.1.概述
Ignite具有若干默认模式并支持创建自定义模式。
默认两个模式可用
SYS模式其中包含许多和集群各种信息有关的系统视图不能在此模式中创建表更多信息请参见系统视图章节的介绍PUBLIC模式未指定模式时的默认模式。
在以下场景中可以创建自定义模式
可以在集群配置中指定自定义模式请参见自定义模式Ignite为通过编程接口或XML配置创建的每个缓存创建一个模式具体请参见缓存和模式名。
#2.2.PUBLIC模式
如果需要且未指定模式时默认会使用PUBLIC模式。例如当通过JDBC接入集群而未显式设置模式时就会使用PUBLIC模式。
#2.3.自定义模式
可以通过IgniteConfiguration的sqlSchemas属性设置自定义模式启动集群之前在配置中指定模式列表然后在运行时就可以在这些模式中创建对象。
下面是带有两个自定义模式的配置示例
XMLJavaC#/.NET
IgniteConfiguration cfg new IgniteConfiguration();SqlConfiguration sqlCfg new SqlConfiguration();sqlCfg.setSqlSchemas(MY_SCHEMA, MY_SECOND_SCHEMA );cfg.setSqlConfiguration(sqlCfg);要接入指定的模式比如通过JDBC驱动那么可以在连接串中指定模式名
jdbc:ignite:thin://127.0.0.1/MY_SCHEMA#2.4.缓存和模式名
当创建带有可查询字段的缓存时可以通过SQL API来对缓存的数据进行维护在SQL层面每个缓存对应一个独立的模式模式的名字等同于缓存的名字。
简单来说当通过SQL API创建了一个表可以通过编程接口将其当做键-值缓存访问而对应的缓存名可以通过CREATE TABLE语句的WITH子句中的CACHE_NAME参数进行指定
CREATE TABLE City (ID INT(11),Name CHAR(35),CountryCode CHAR(3),District CHAR(20),Population INT(11),PRIMARY KEY (ID, CountryCode)
) WITH backups1, CACHE_NAMECity;更多信息请参见CREATE TABLE章节的介绍。
如果未指定这个参数缓存名定义为如下格式大写格式
SQL_SCHEMA_NAME_TABLE_NAME#3.定义索引
除了常规的DDL命令比如CREATE/DROP INDEX开发者还可以使用SQL API来定义索引。
提示
索引的功能是通过ignite-indexing模块提供的所以如果通过Java代码启动Ignite需要将这个模块加入类路径。
Ignite会自动为每个缓存的主键和关联键字段创建索引当在值对象的字段上创建索引时Ignite会创建一个由索引字段和主键字段组成的组合索引。在SQL的角度该索引由2列组成索引列和主键列。
#3.1.使用SQL创建索引
具体请参见CREATE INDEX章节的内容。
#3.2.使用注解配置索引
索引和可查询字段在代码上可以通过QuerySqlField注解进行配置。在下面的示例中Ignite的SQL引擎会在id和salary字段上创建索引
JavaC#/.NET
public class Person implements Serializable {/** Indexed field. Will be visible to the SQL engine. */QuerySqlField(index true)private long id;/** Queryable field. Will be visible to the SQL engine. */QuerySqlFieldprivate String name;/** Will NOT be visible to the SQL engine. */private int age;/*** Indexed field sorted in descending order. Will be visible to the SQL engine.*/QuerySqlField(index true, descending true)private float salary;
}SQL查询中类型名会被用作表名这时表名为Person使用的模式名和定义见模式章节的介绍。
id和salary都是索引字段id为升序排列而salary为倒序排列。
如果不希望索引一个字段但是希望在SQL查询中使用该列那么该字段需要加上该注解但是不需要index true参数这样的字段叫做可查询字段在上例中name定义为可查询字段。
age字段既不是可查询字段也不是一个索引字段因此在SQL查询中是无法访问的。
定义索引字段后还需要注册索引类型。
运行时更新索引和可查询字段
如果希望运行时管理索引或者对象字段的可见性需要使用CREATE/DROP INDEX命令。
#3.2.1.索引嵌套对象
使用注解对象的嵌套字段也可以被索引和查询。比如考虑一个Person对象内部有一个Address对象
public class Person {/** Indexed field. Will be visible for SQL engine. */QuerySqlField(index true)private long id;/** Queryable field. Will be visible for SQL engine. */QuerySqlFieldprivate String name;/** Will NOT be visible for SQL engine. */private int age;/** Indexed field. Will be visible for SQL engine. */QuerySqlField(index true)private Address address;
}而Address类的结构如下
public class Address {/** Indexed field. Will be visible for SQL engine. */QuerySqlField (index true)private String street;/** Indexed field. Will be visible for SQL engine. */QuerySqlField(index true)private int zip;
}在上面的示例中Address类的所有字段都加上了QuerySqlField(index true)注解Person类的Address对象也加上了该注解。
这样就可以执行下面的SQL语句
QueryCursorList? cursor personCache.query(new SqlFieldsQuery( select * from Person where street street1));注意在SQL语句的WHERE条件中不需要指定address.street这是因为Address类的字段会被合并到Person中这样就可以简单地在查询中直接访问Address中的字段。
警告
如果在嵌套对象上创建了索引就不能在这个表上执行UPDATE或者INSERT语句。
#3.2.2.注册索引类型
定义索引和可查询字段之后需要将它们及其所属的对象类型一起注册到SQL引擎中。
要指定应建立索引的类型需要在CacheConfiguration.setIndexedTypes()方法中传递相应的键-值对如下例所示
JavaC#/.NET
// Preparing configuration.
CacheConfigurationLong, Person ccfg new CacheConfiguration();// Registering indexed type.
ccfg.setIndexedTypes(Long.class, Person.class);此方法仅接受成对的类型一个键类一个值类基本类型需要用包装器类型传入。
预定义字段
除了用QuerySqlField注解标注的所有字段每个表都有两个特别的预定义字段_key和_val它表示到整个键对象和值对象的引用。这很有用比如当它们中的一个是基本类型并且希望用它的值进行过滤时执行SELECT * FROM Person WHERE _key 100查询即可。
注意
因为有二进制编组器不需要将索引类型类加入集群节点的类路径中SQL引擎不需要对象反序列化就可以钻取索引和可查询字段的值。
#3.2.3.组合索引
当查询条件复杂时可以使用多字段索引来加快查询的速度这时可以用QuerySqlField.Group注解。如果希望一个字段参与多个组合索引时也可以将多个QuerySqlField.Group注解加入orderedGroups中。
比如下面的Person类中age字段加入了名为age_salary_idx的组合索引它的分组序号是0并且降序排列同一个组合索引中还有一个字段salary,它的分组序号是3并且升序排列。最重要的是salary字段还是一个单列索引(除了orderedGroups声明之外还加上了index true)。分组中的order不需要是什么特别的数值它只是用于分组内的字段排序。
JavaC#/.NET
public class Person implements Serializable {/** Indexed in a group index with salary. */QuerySqlField(orderedGroups { QuerySqlField.Group(name age_salary_idx, order 0, descending true) })private int age;/** Indexed separately and in a group index with age. */QuerySqlField(index true, orderedGroups { QuerySqlField.Group(name age_salary_idx, order 3) })private double salary;
}注意
将QuerySqlField.Group放在QuerySqlField(orderedGroups{...})外面是无效的。
#3.3.使用查询实体配置索引
索引和字段也可以通过org.apache.ignite.cache.QueryEntity进行配置它便于利用Spring进行基于XML的配置。
在上面基于注解的配置中涉及的所有概念对于基于QueryEntity的方式也都有效此外如果类型的字段通过QuerySqlField进行了配置并且通过CacheConfiguration.setIndexedTypes注册过的在内部也会被转换为查询实体。
下面的示例显示的是如何定义单一字段索引、组合索引和可查询字段
XMLJavaC#/.NET
CacheConfigurationLong, Person cache new CacheConfigurationLong, Person(myCache);QueryEntity queryEntity new QueryEntity();queryEntity.setKeyFieldName(id).setKeyType(Long.class.getName()).setValueType(Person.class.getName());LinkedHashMapString, String fields new LinkedHashMap();
fields.put(id, java.lang.Long);
fields.put(name, java.lang.String);
fields.put(salary, java.lang.Long);queryEntity.setFields(fields);queryEntity.setIndexes(Arrays.asList(new QueryIndex(name),new QueryIndex(Arrays.asList(id, salary), QueryIndexType.SORTED)));cache.setQueryEntities(Arrays.asList(queryEntity));SQL查询中会使用valueType的简称作为表名这时表名为Person模式名的用法和定义请参见理解模式章节的内容。
QueryEntity定义之后就可以执行下面的查询了
SqlFieldsQuery qry new SqlFieldsQuery(SELECT id, name FROM Person WHERE id 1500 LIMIT 10);运行时更新索引和可查询字段
如果希望运行时管理索引或者对象字段的可见性需要使用CREATE/DROP INDEX命令。
#3.4.配置索引内联值
正确的索引内联值有助于增加索引字段上的查询速度关于如何选择正确的内联值请参见增加索引内联值章节的介绍。
大多数情况下只需要为可变长度字段的索引设置内联值比如字符串或者数组默认值是10。
可通过如下方式修改默认值
单独为每个索引配置内联值通过CacheConfiguration.sqlIndexMaxInlineSize属性为缓存内的所有索引配置内联值通过IGNITE_MAX_INDEX_PAYLOAD_SIZE系统属性为集群内的所有索引配置内联值。
配置将按照上面的顺序依次生效。
可以为每个索引单独配置内联值这会覆盖默认值。如果要为开发者定义的索引设置内联值可以用下面的方法之一该值以字节数为单位。
注解方式
JavaC#/.NET
QuerySqlField(index true, inlineSize 13)
private String country;QueryEntity方式
JavaC#/.NET
QueryIndex idx new QueryIndex(country);
idx.setInlineSize(13);
queryEntity.setIndexes(Arrays.asList(idx));CREATE INDEX命令
如果使用的是CREATE INDEX命令那么可以使用INLINE_SIZE选项来配置内联值
create index country_idx on Person (country) INLINE_SIZE 13;#3.5.自定义键
如果只使用预定义的SQL数据类型作为缓存键那么就没必要对和DML相关的配置做额外的操作这些数据类型在GridQueryProcessor#SQL_TYPES常量中进行定义列举如下
所有的基本类型及其包装器除了char和CharacterString;BigDecimal;byte[];java.util.Date, java.sql.Date, java.sql.Timestamp;java.util.UUID。
不过如果决定引入复杂的自定义缓存键那么在DML语句中要指向这些字段就需要
在QueryEntity中定义这些字段与在值对象中配置字段一样使用新的配置参数QueryEntitty.setKeyFields(..)来对键和值进行区分。
下面的例子展示了如何实现
XMLJavaC#/.NET
// Preparing cache configuration.
CacheConfigurationLong, Person cacheCfg new CacheConfigurationLong, Person(personCache);// Creating the query entity.
QueryEntity entity new QueryEntity(CustomKey, Person);// Listing all the queryable fields.
LinkedHashMapString, String fields new LinkedHashMap();fields.put(intKeyField, Integer.class.getName());
fields.put(strKeyField, String.class.getName());fields.put(firstName, String.class.getName());
fields.put(lastName, String.class.getName());entity.setFields(fields);// Listing a subset of the fields that belong to the key.
SetString keyFlds new HashSet();keyFlds.add(intKeyField);
keyFlds.add(strKeyField);entity.setKeyFields(keyFlds);// End of new settings, nothing else here is DML relatedentity.setIndexes(Collections.QueryIndexemptyList());cacheCfg.setQueryEntities(Collections.singletonList(entity));ignite.createCache(cacheCfg);哈希值自动计算和equals实现
如果自定义键可以被序列化为二进制形式那么Ignite会自动进行哈希值的计算并且实现equals方法。
但是如果键类型是Externalizable类型那么就无法序列化为二进制形式那么就需要自行实现hashCode和equals方法具体请参见使用二进制对象章节的介绍。
#4.使用SQL API
除了使用JDBC驱动Java开发者还可以使用Ignite的SQL API来访问和修改Ignite中存储的数据。
SqlFieldsQuery类是执行SQL查询和处理结果集的接口SqlFieldsQuery通过IgniteCache.query(SqlFieldsQuery)方法执行然后会返回一个游标。
#4.1.配置可查询字段
如果希望使用SQL语句来查询缓存需要定义值对象的哪些字段是可查询的可查询字段是数据模型中SQL引擎可以处理的字段。
提示
如果使用JDBC或者SQL工具建表则不需要定义可查询字段。
提示
索引的功能是通过ignite-indexing模块提供的所以如果通过Java代码启动Ignite需要将这个模块加入类路径。
在Java中可查询字段可以通过两种方式来定义
使用注解通过查询实体定义。
#4.1.1.QuerySqlField注解
要让某个字段可查询需要在值类定义的对应字段上加注QuerySqlField注解然后调用CacheConfiguration.setIndexedTypes(…)方法。
JavaC#/.NET
class Person implements Serializable {/** Indexed field. Will be visible to the SQL engine. */QuerySqlField(index true)private long id;/** Queryable field. Will be visible to the SQL engine. */QuerySqlFieldprivate String name;/** Will NOT be visible to the SQL engine. */private int age;/*** Indexed field sorted in descending order. Will be visible to the SQL engine.*/QuerySqlField(index true, descending true)private float salary;
}public static void main(String[] args) {Ignite ignite Ignition.start();CacheConfigurationLong, Person personCacheCfg new CacheConfigurationLong, Person();personCacheCfg.setName(Person);personCacheCfg.setIndexedTypes(Long.class, Person.class);IgniteCacheLong, Person cache ignite.createCache(personCacheCfg);
}#4.1.2.查询实体
可以通过QueryEntity类来定义可查询字段查询实体可以通过XML来配置
XMLJavaC#/.NET
class Person implements Serializable {private long id;private String name;private int age;private float salary;
}public static void main(String[] args) {Ignite ignite Ignition.start();CacheConfigurationLong, Person personCacheCfg new CacheConfigurationLong, Person();personCacheCfg.setName(Person);QueryEntity queryEntity new QueryEntity(Long.class, Person.class).addQueryField(id, Long.class.getName(), null).addQueryField(age, Integer.class.getName(), null).addQueryField(salary, Float.class.getName(), null).addQueryField(name, String.class.getName(), null);queryEntity.setIndexes(Arrays.asList(new QueryIndex(id), new QueryIndex(salary, false)));personCacheCfg.setQueryEntities(Arrays.asList(queryEntity));IgniteCacheLong, Person cache ignite.createCache(personCacheCfg);
}#4.2.查询
要在缓存上执行查询简单地创建一个SqlFieldsQuery对象将查询字符串传给构造方法然后执行cache.query(…)即可。注意在下面的示例中Person缓存必须配置为对SQL引擎可见。
JavaC#/.NETC
IgniteCacheLong, Person cache ignite.cache(Person);SqlFieldsQuery sql new SqlFieldsQuery(select concat(firstName, , lastName) from Person);// Iterate over the result set.
try (QueryCursorList? cursor cache.query(sql)) {for (List? row : cursor)System.out.println(personName row.get(0));
}SqlFieldsQuery会返回一个游标然后可以用游标来迭代匹配SQL查询的结果集。
#4.2.1.本地执行
如果要强制一个查询在本地执行可以使用SqlFieldsQuery.setLocal(true)方法。这时查询是在执行查询的节点的本地数据上执行这意味着查询的结果集是不完整的所以使用这个模式前要了解这个限制。
#4.2.2.WHERE子句的子查询
INSERT、MERGE语句中的SELECT查询以及由UPDATE和DELETE操作生成的SELECT查询也是分布式的可以以并置或非并置的模式执行。
但是如果WHERE子句中有一个子查询那么其只能以并置的方式执行。
比如考虑下面的查询
DELETE FROM Person WHERE id IN(SELECT personId FROM Salary s WHERE s.amount 2000);SQL引擎会生成一个SELECT查询来获取要删除的条目列表。该查询是分布式的在整个集群中执行大致如下
SELECT _key, _val FROM Person WHERE id IN(SELECT personId FROM Salary s WHERE s.amount 2000);但是IN子句中的子查询SELECT personId FROM Salary …并不是分布式的只能在节点的本地可用数据集上执行。
#4.3.插入、更新、删除和合并
使用SqlFieldsQuery可以执行DML命令来修改数据
INSERTUPDATEDELETEMERGE
IgniteCacheLong, Person cache ignite.cache(personCache);cache.query(new SqlFieldsQuery(INSERT INTO Person(id, firstName, lastName) VALUES(?, ?, ?)).setArgs(1L, John, Smith)).getAll();当使用SqlFieldsQuery来执行DDL语句时必须调用query(…)方法返回的游标的getAll()方法。
#4.4.指定模式
通过SqlFieldsQuery执行的任何SELECT语句默认都是在PUBLIC模式下解析的。但是如果表不在这个模式下需要调用SqlFieldsQuery.setSchema(…)来指定模式这样语句就在指定的模式下执行了。
JavaC#/.NETC
SqlFieldsQuery sql new SqlFieldsQuery(select name from City).setSchema(PERSON);另外也可以在语句中指定模式
SqlFieldsQuery sql new SqlFieldsQuery(select name from Person.City);#4.5.创建表
可以向SqlFieldsQuery传递任何受支持的DDL语句如下所示
JavaC#/.NETC
IgniteCacheLong, Person cache ignite.getOrCreateCache(new CacheConfigurationLong, Person().setName(Person));// Creating City table.
cache.query(new SqlFieldsQuery(CREATE TABLE City (id int primary key, name varchar, region varchar))).getAll();在SQL模式方面上述代码的执行结果创建了下面的表
Person模式中的Person表如果之前未创建Person模式中的City表。
要查询City表可以使用两种方式select * from Person.City或new SqlFieldsQuery(select * from City).setSchema(PERSON)注意大写。
#4.6.取消查询
有两种方式可以取消长时间运行的查询。
第一种方式是设置查询执行超时
JavaC#/.NET
SqlFieldsQuery query new SqlFieldsQuery(SELECT * from Person);// Setting query execution timeout
query.setTimeout(10_000, TimeUnit.SECONDS);第二个方式是调用QueryCursor.close()来终止查询
JavaC#/.NET
SqlFieldsQuery query new SqlFieldsQuery(SELECT * FROM Person);// Executing the query
QueryCursorList? cursor cache.query(query);// Halting the query that might be still in progress.
cursor.close();#4.7.示例
Ignite的源代码中有一个直接可以运行的SqlDmlExample其演示了所有上面提到过的DML操作的使用。
#5.分布式关联
分布式关联是指SQL语句中通过关联子句组合了两个或者更多的分区表如果这些表关联在分区列关联键上该关联称为并置关联否则称为非并置关联。
并置关联更高效因为其可以高效地在集群节点间分布。
Ignite默认将每个关联查询都视为并置关联并按照并置的模式执行。
警告
如果查询是非并置的需要通过SqlFieldsQuery.setDistributedJoins(true)来开启查询执行的非并置模式否则查询的结果集会是不正确的。
警告
如果经常关联表那么建议将表在同一个列关联表的列上进行分区。
非并置的关联仅适用于无法使用并置关联的场景。
#5.1.并置关联
下图解释了并置关联的执行过程一个并置关联Q会被发给存储与查询条件匹配的数据的所有节点然后查询在每个节点的本地数据集上执行E(Q)结果集R会在查询的发起节点客户端节点聚合 #5.2.非并置关联
如果以非并置模式执行查询则SQL引擎将在存储与查询条件匹配的数据的所有节点上本地执行查询。但是因为数据不是并置的所以每个节点将通过发送广播或单播请求从其他节点拉取缺失的数据本地不存在下图描述了此过程 如果关联是在主键或关联键上则节点将发送单播请求因为这时节点知道缺失数据的位置。否则节点将发送广播请求。出于性能原因广播和单播请求都被汇总为批次。
通过设置JDBC/ODBC参数或通过调用SqlFieldsQuery.setDistributedJoins(true)使用SQL API可以启用非并置查询执行模式。
警告
如果对复制表中的列使用非并置关联则该列必须有索引。否则会抛出异常。
#6.SQL事务
警告
支持SQL事务当前处于测试阶段生产环境建议使用键-值事务。
#6.1.概述
配置为TRANSACTIONAL_SNAPSHOT原子化模式的缓存支持SQL事务。TRANSACTIONAL_SNAPSHOT模式是Ignite缓存的多版本并发控制MVCC实现关于MVCC的更多信息以及当前的限制请参见多版本并发控制章节的内容。
关于Ignite支持的事务语法请参见事务章节的内容。
#6.2.启用MVCC
在缓存配置中使用TRANSACTIONAL_SNAPSHOT原子化模式可以为缓存开启MVCC如果使用CREATE TABLE命令建表可以在命令的WITH子句中指定原子化模式参数。
XMLJavaC#/.NETSQL
CacheConfiguration cacheCfg new CacheConfiguration();
cacheCfg.setName(myCache);cacheCfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT);#6.3.限制
#6.3.1.跨缓存事务
TRANSACTIONAL_SNAPSHOT模式是缓存级的因此不允许在一个事务中的缓存具有不同的原子化模式如果要在一个事务中覆盖多张表那么所有的相关表都要使用TRANSACTIONAL_SNAPSHOT模式创建。
#6.3.2.嵌套事务
通过JDBC/ODBC连接参数Ignite支持三种模式用于处理嵌套的SQL事务。
JDBC连接串示例
jdbc:ignite:thin://127.0.0.1/?nestedTransactionsModeCOMMIT当事务中发生了嵌套的事务系统的行为取决于nestedTransactionsMode参数
ERROR如果遇到嵌套事务会抛出错误并且包含的事务会回滚这是默认的行为COMMIT包含事务会被挂起嵌套事务启动后如果遇到COMMIT语句会被提交。包含事务中的其余语句会作为隐式事务执行IGNORE不要使用这个模式嵌套事务的开始会被忽略嵌套事务中的语句会作为包含事务的一部分执行并且随着嵌套事务的提交而提交所有的变更包含事务的剩余语句会作为隐式事务执行。
#7.自定义SQL函数
Ignite的SQL引擎支持通过额外用Java编写的自定义SQL函数来扩展ANSI-99规范定义的SQL函数集。
一个自定义SQL函数仅仅是一个加注了QuerySqlFunction注解的公共静态方法。
// Defining a custom SQL function.
public class MyFunctions {QuerySqlFunctionpublic static int sqr(int x) {return x * x;}
}持有自定义SQL函数的类需要使用setSqlFunctionClasses(...)方法在某个CacheConfiguration中注册。
// Preparing a cache configuration.
CacheConfiguration cfg new CacheConfiguration();// Registering the class that contains custom SQL functions.
cfg.setSqlFunctionClasses(MyFunctions.class);经过了上述配置的缓存部署之后在SQL查询中就可以调用自定义函数了如下所示
// Preparing the query that uses customly defined sqr function.
SqlFieldsQuery query new SqlFieldsQuery(SELECT name FROM Blocks WHERE sqr(size) 100);// Executing the query.
cache.query(query).getAll();类注册
在自定义SQL函数可能要执行的所有节点上通过CacheConfiguration.setSqlFunctionClasses(...)注册的类都需要添加到类路径中否则在自定义函数执行时会抛出ClassNotFoundException异常。
#8.JDBC驱动
Ignite提供了JDBC驱动可以通过标准的SQL语句处理分布式数据比如从JDBC端直接进行SELECT、INSERT、UPDATE和DELETE。
目前Ignite支持两种类型的驱动轻量易用的JDBC Thin模式驱动以及以客户端节点形式与集群进行交互的JDBC客户端驱动。
#8.1.JDBC Thin模式驱动
JDBC Thin模式驱动是Ignite提供的默认轻量级驱动要使用这种驱动只需要将ignite-core-{version}.jar加入应用的类路径即可。
驱动会接入集群的一个节点然后将所有的请求转发给它进行处理。节点会处理分布式的查询以及结果集的汇总然后将结果集反馈给客户端应用。
JDBC连接串可以有两种模式URL查询模式以及分号模式
// URL query pattern
jdbc:ignite:thin://hostAndPortRange0[,hostAndPortRange1]...[,hostAndPortRangeN][/schema][?params]hostAndPortRange : host[:port_from[..port_to]]params : param1value1[param2value2]...[paramNvalueN]// Semicolon pattern
jdbc:ignite:thin://hostAndPortRange0[,hostAndPortRange1]...[,hostAndPortRangeN][;schemaschema_name][;param1value1]...[;paramNvalueN]host必需它定义了要接入的集群节点主机地址port_from打开连接的端口范围的起始点如果忽略此参数默认为10800port_to可选如果忽略此参数则等同于port_fromschema要访问的模式名默认是PUBLIC这个名字对应于SQL的ANSI-99标准不加引号是大小写不敏感的加引号是大小写敏感的。如果使用了分号模式模式可以通过参数名schema定义params可选。
驱动类名为org.apache.ignite.IgniteJdbcThinDriver比如下面就是如何打开到集群节点的连接监听地址为192.168.0.50
// Register JDBC driver.
Class.forName(org.apache.ignite.IgniteJdbcThinDriver);// Open the JDBC connection.
Connection conn DriverManager.getConnection(jdbc:ignite:thin://192.168.0.50);如果通过bash接入则JDBC URL需要加引号
如果通过bash环境接入则连接URL需要加 比如jdbc:ignite:thin://[address]:[port];user[username];password[password]
#8.1.1.参数
下表列出了JDBC连接串支持的所有参数
属性名描述默认值userSQL连接的用户名如果服务端开启了认证则此参数为必需。关于如何开启认证和创建用户可以分别参见认证和创建用户的文档。ignitepasswordSQL连接的密码如果服务端开启了认证则此参数为必需。关于如何开启认证和创建用户可以分别参见认证和创建用户的文档。ignitedistributedJoins对于非并置数据是否使用分布式关联falseenforceJoinOrder是否在查询中强制表的关联顺序如果配置为true查询优化器在关联中不会对表进行重新排序。falsecollocated如果SQL语句包含按主键或关联键对结果集进行分组的GROUP BY子句可以将此参数设置为true。当Ignite执行分布式查询时会向单个集群节点发送子查询如果事先知道待查询的数据是在同一个节点上并置在一起的并且是按主键或关联键分组的那么Ignite通过在参与查询的每个节点本地分组数据来实现显著的性能和网络优化。falsereplicatedOnly查询是否只包含复制表这是一个潜在的可能提高性能的提示。falseautoCloseServerCursor当拿到最后一个结果集时是否自动关闭服务端游标。开启之后对ResultSet.close()的调用就不需要网络访问这样会改进性能。但是如果服务端游标已经关闭在调用ResultSet.getMetadata()方法时会抛出异常这时为什么默认值为false的原因。falsepartitionAwareness启用分区感知模式该模式中驱动会尝试确定要查询的数据所在的节点然后把请求发给这些节点。falsepartitionAwarenessSQLCacheSize驱动为优化而在本地保留的不同SQL查询数。当第一次执行查询时驱动会接收正在查询的表的分区分布并将其保存以备将来在本地使用。下次查询此表时驱动使用该分区分布来确定要查询的数据的位置以便将查询直接发送到正确的节点。当集群拓扑发生变更时此包含SQL查询的本地存储将失效。此参数的最佳值应等于要执行的不同SQL查询的数量。1000partitionAwarenessPartitionDistributionsCacheSize表示分区分布的不同对象的数量驱动在本地保留以进行优化。具体请参见partitionAwarenessSQLCacheSize参数的说明。当集群拓扑发生变更时持有分区分布对象的本地存储将失效。此参数的最佳值应等于要在查询中使用的不同表缓存组的数量。1000socketSendBuffer发送套接字缓冲区大小如果配置为0会使用操作系统默认值。0socketReceiveBuffer接收套接字缓冲区大小如果配置为0会使用操作系统默认值。0tcpNoDelay是否使用TCP_NODELAY选项。truelazy查询延迟执行。Ignite默认会将所有的结果集放入内存然后将其返回给客户端。对于不太大的结果集这样会提供较好的性能并且使内部的数据库锁时间最小化因此提高了并发能力。但是如果相对于可用内存来说结果集过大那么会导致频繁的GC暂停甚至OutOfMemoryError如果使用这个标志可以提示Ignite延迟加载结果集这样可以在不大幅降低性能的前提下最大限度地减少内存的消耗。falseskipReducerOnUpdate开启服务端的更新特性。当Ignite执行DML操作时首先它会获取所有受影响的中间行给查询发起方进行分析通常被称为汇总方然后会准备一个更新值的批次发给远程节点。这个方式可能影响性能如果一个DML操作需要移动大量数据时还可能会造成网络堵塞。使用这个标志可以提示Ignite在对应的远程节点上进行中间行的分析和更新。默认值为false这意味着会首先获取中间行然后发给查询发起方。false
关于和安全有关的参数请参见使用SSL章节的介绍。
#8.1.2.连接串示例
jdbc:ignite:thin://myHost接入myHost,其它比如端口为10800等都是默认值jdbc:ignite:thin://myHost:11900接入myHost,自定义端口为11900其它为默认值jdbc:ignite:thin://myHost:11900;userignite;passwordignite接入myHost,自定义端口为11900并且带有用于认证的用户凭据jdbc:ignite:thin://myHost:11900;distributedJoinstrueautoCloseServerCursortrue接入myHost,自定义端口为11900开启了分布式关联和autoCloseServerCursor优化jdbc:ignite:thin://myHost:11900/myschema;接入myHost自定义端口为11900模式为MYSCHEMAjdbc:ignite:thin://myHost:11900/MySchema;lazyfalse接入myHost自定义端口为11900模式为MySchema模式名区分大小写并且禁用了查询的延迟执行。
#8.1.3.多端点
在连接串中配置多个连接端点也是可以的这样如果连接中断会开启自动故障转移JDBC驱动会从列表中随机选择一个地址接入。如果之前的连接中断驱动会选择另一个地址直到连接恢复如果所有的端点都不可达JDBC会停止重连并且抛出异常。
下面的示例会显示如何通过连接串传递3个地址
// Register JDBC driver.
Class.forName(org.apache.ignite.IgniteJdbcThinDriver);// Open the JDBC connection passing several connection endpoints.
Connection conn DriverManager.getConnection(jdbc:ignite:thin://192.168.0.50:101,192.188.5.40:101, 192.168.10.230:101);#8.1.4.分区感知
警告
分区感知是一个试验性特性API和设计架构在正式发布之前可能会变更。
分区感知是一个可使JDBC驱动“感知”集群中分区分布的功能。它使得驱动可以选择持有待查询数据的节点并将查询直接发送到那些节点如果在驱动的配置中提供了节点的地址。分区感知可以提高使用关联键的查询的平均性能。
没有分区感知时JDBC驱动将连接到某个节点然后所有查询都通过该节点执行。如果数据分布在其他节点上则必须在集群内重新路由查询这会增加一个额外的网络波动。分区感知通过将查询直接发送到正确的节点来消除该波动。
要使用分区感知功能需要在连接属性中提供所有服务端节点的地址驱动会将请求直接发送到存储查询所请求数据的节点。
警告
注意当前需要在连接属性中提供所有服务端节点的地址因为在打开连接后驱动不会自动加载它们。这意味着如果新的服务端节点加入集群需要将节点的地址添加到连接属性中然后重新连接驱动否则驱动将无法直接向该节点发送请求。
要开启分区感知需要将partitionAwarenesstrue参数添加到连接串中然后提供多个服务端节点的地址。
Class.forName(org.apache.ignite.IgniteJdbcThinDriver);Connection conn DriverManager.getConnection(jdbc:ignite:thin://192.168.0.50,192.188.5.40,192.168.10.230?partitionAwarenesstrue);提示
分区感知功能只能使用默认的关联函数。
#8.1.5.集群配置
为了接收和处理来自JDBC Thin驱动转发过来的请求一个节点需要绑定到一个本地网络端口10800然后监听入站请求。
通过ClientConnectorConfiguration可以对参数进行修改
JavaXML
IgniteConfiguration cfg new IgniteConfiguration().setClientConnectorConfiguration(new ClientConnectorConfiguration());其支持如下的参数
参数名描述默认值host绑定的主机名或者IP地址如果配置为null会使用localHost。nullport绑定的TCP端口如果指定的端口已被占用Ignite会使用portRange属性来查找其它可用的端口。10800portRange定义尝试绑定的端口数量比如如果端口配置为10800并且端口范围为100Ignite会从10800开始在[10800,10900]范围内查找可用端口。100maxOpenCursorsPerConnection每个连接打开的服务端游标的最大数量。128threadPoolSize线程池中负责请求处理的线程数量。max(8,CPU核数)socketSendBufferSizeTCP套接字发送缓冲区大小如果配置为0会使用操作系统默认值。0socketReceiveBufferSizeTCP套接字接收缓冲区大小如果配置为0会使用操作系统默认值。0tcpNoDelay是否使用TCP_NODELAY选项。trueidleTimeout客户端连接空闲超时时间。在空闲超过配置的超时时间后客户端与服务端的连接会断开。如果该参数配置为0或者负值空闲超时会被禁用。0isJdbcEnabled是否允许JDBC访问。trueisThinClientEnabled是否允许瘦客户端访问。truesslEnabled如果开启SSL只允许SSL客户端连接。一个节点只允许一种连接模式SSL或普通一个节点无法同时接收两种模式的客户端连接但是这个参数集群中的各个节点可以不同。falseuseIgniteSslContextFactory在Ignite配置中是否使用SSL上下文工厂具体可以看IgniteConfiguration.sslContextFactory。truesslClientAuth是否需要客户端认证。falsesslContextFactory提供节点侧SSL的FactorySSLContext实现的类名。null
JDBC Thin模式驱动并非线程安全
JDBC对象中的Connection、Statement和ResultSet不是线程安全的。因此不能在多线程中使用一个JDBC连接的Statement和ResultSet。 JDBC Thin模式驱动防止并发如果检测到了并发访问那么会抛出SQLException消息为Concurrent access to JDBC connection is not allowed [ownThreadguard_owner_thread_name,curThreadcurrent_thread_name],SQLSTATE08006。
#8.1.6.使用SSL
JDBC Thin模式驱动可以使用SSL来保护与集群之间的通信集群端和驱动端必须同时配置SSL集群配置方面请参见瘦客户端和JDBC/ODBC的SSL/TLS章节的介绍。
JDBC驱动中开启SSL需要在连接串中传递sslModerequire参数并且提供密钥库和信任库参数
Class.forName(org.apache.ignite.IgniteJdbcThinDriver);String keyStore keystore/node.jks;
String keyStorePassword 123456;String trustStore keystore/trust.jks;
String trustStorePassword 123456;try (Connection conn DriverManager.getConnection(jdbc:ignite:thin://127.0.0.1?sslModerequire sslClientCertificateKeyStoreUrl keyStore sslClientCertificateKeyStorePassword keyStorePassword sslTrustCertificateKeyStoreUrl trustStore sslTrustCertificateKeyStorePassword trustStorePassword)) {ResultSet rs conn.createStatement().executeQuery(select 10);rs.next();System.out.println(rs.getInt(1));
} catch (Exception e) {e.printStackTrace();
}下表列出了和SSL/TLS连接有关的参数
参数名描述默认值sslMode开启SSL连接。可用的模式为1.require在客户端开启SSL协议只有SSL连接才可以接入。2.disable在客户端禁用SSL协议只支持普通连接。disablesslProtocol安全连接的协议名如果未指定会使用TLS协议。协议实现由JSSE提供SSLv3 (SSL), TLSv1 (TLS), TLSv1.1, TLSv1.2TLSsslKeyAlgorithm用于创建密钥管理器的密钥管理器算法。注意多数情况使用默认值即可。算法实现由JSSE提供PKIX (X509或SunPKIX), SunX509sslClientCertificateKeyStoreUrl客户端密钥存储库文件的url这是个强制参数因为没有密钥管理器SSL上下文无法初始化。如果sslMode为require并且未通过属性文件指定密钥存储库 URL那么会使用JSSE属性javax.net.ssl.keyStore的值。JSSE系统属性javax.net.ssl.keyStore的值。sslClientCertificateKeyStorePassword客户端密钥存储库密码。如果sslMode为require并且未通过属性文件指定密钥存储库密码那么会使用JSSE属性javax.net.ssl.keyStorePassword的值。JSSE属性javax.net.ssl.keyStorePassword的值。sslClientCertificateKeyStoreType用于上下文初始化的客户端密钥存储库类型。如果sslMode为require并且未通过属性文件指定密钥存储库类型那么会使用JSSE属性javax.net.ssl.keyStoreType的值。JSSE属性javax.net.ssl.keyStoreType的值如果属性未定义默认值为JKS。sslTrustCertificateKeyStoreUrltruststore文件的URL。这是个可选参数但是sslTrustCertificateKeyStoreUrl和sslTrustAll必须配置一个。如果sslMode为require并且未通过属性文件指定truststore文件URL那么会使用JSSE属性javax.net.ssl.trustStore的值。JSSE系统属性javax.net.ssl.trustStore的值。sslTrustCertificateKeyStorePasswordtruststore密码。如果sslMode为require并且未通过属性文件指定truststore密码那么会使用JSSE属性javax.net.ssl.trustStorePassword的值。JSSE系统属性javax.net.ssl.trustStorePassword的值。sslTrustCertificateKeyStoreTypetruststore类型。如果sslMode为require并且未通过属性文件指定truststore类型那么会使用JSSE属性javax.net.ssl.trustStoreType的值。JSSE系统属性javax.net.ssl.trustStoreType的值。如果属性未定义默认值为JKS。sslTrustAll禁用服务端的证书验证。配置为true信任任何服务端证书撤销的、过期的或者自签名的SSL证书。注意如果不能完全信任网络比如公共互联网不要在生产中启用该选项。falsesslFactoryFactorySSLSocketFactory的自定义实现的类名如果sslMode为require并且指定了该工厂类自定义的工厂会替换JSSE的默认值这时其它的SSL属性也会被忽略。null
默认实现基于JSSE并且需要处理两个Java密钥库文件。
sslClientCertificateKeyStoreUrl客户端认证密钥库文件其持有客户端的密钥和证书sslTrustCertificateKeyStoreUrl可信证书密钥库文件包含用于验证服务器证书的证书信息。
信任库是可选参数但是sslTrustCertificateKeyStoreUrl或者sslTrustAll必须配置两者之一。
使用sslTrustAll参数
如果生产环境位于不完全可信网络尤其是公共互联网不要开启此选项。
如果希望使用自己的实现或者通过某种方式配置SSLSocketFactory可以使用驱动的sslFactory参数这是一个包含FactorySSLSocketFactory接口实现的类名字符串该类对于JDBC驱动的类加载器必须可用。
#8.2.Ignite DataSource
DataSource对象可用作部署对象其可以通过JNDI命名服务按逻辑名定位。Ignite JDBC驱动的org.apache.ignite.IgniteJdbcThinDataSource实现了JDBC的DataSource接口这样就可以使用DataSource接口了。
除了通用的DataSource属性外IgniteJdbcThinDataSource还支持所有可以传递给JDBC连接字符串的Ignite特有属性。例如distributedJoins属性可以通过IgniteJdbcThinDataSource#setDistributedJoins()方法进行调整。
具体请参见IgniteJdbcThinDataSource的javadoc。
#8.3.示例
要处理集群中的数据需要使用下面的一种方式来创建一个JDBCConnection对象
// Open the JDBC connection via DriverManager.
Connection conn DriverManager.getConnection(jdbc:ignite:thin://192.168.0.50);或者
// Or open connection via DataSource.
IgniteJdbcThinDataSource ids new IgniteJdbcThinDataSource();
ids.setUrl(jdbc:ignite:thin://127.0.0.1);
ids.setDistributedJoins(true);Connection conn ids.getConnection();之后就可以执行SELECTSQL查询了
// Query people with specific age using prepared statement.
PreparedStatement stmt conn.prepareStatement(select name, age from Person where age ?);stmt.setInt(1, 30);ResultSet rs stmt.executeQuery();while (rs.next()) {String name rs.getString(name);int age rs.getInt(age);// ...
}此外可以使用DML语句对数据进行修改。
#8.3.1.INSERT
// Insert a Person with a Long key.
PreparedStatement stmt conn.prepareStatement(INSERT INTO Person(_key, name, age) VALUES(CAST(? as BIGINT), ?, ?));stmt.setInt(1, 1);
stmt.setString(2, John Smith);
stmt.setInt(3, 25);stmt.execute();#8.3.2.MERGE
// Merge a Person with a Long key.
PreparedStatement stmt conn.prepareStatement(MERGE INTO Person(_key, name, age) VALUES(CAST(? as BIGINT), ?, ?));stmt.setInt(1, 1);
stmt.setString(2, John Smith);
stmt.setInt(3, 25);stmt.executeUpdate();#8.3.3.UPDATE
// Update a Person.
conn.createStatement().executeUpdate(UPDATE Person SET age age 1 WHERE age 25);#8.3.4.DELETE
conn.createStatement().execute(DELETE FROM Person WHERE age 25);#8.4.流处理
Ignite的JDBC驱动可以通过SET STREAMING命令对流化数据进行批量处理具体可以看SET STREAMING的相关内容。
#8.5.错误码
Ignite的JDBC驱动将错误码封装进了java.sql.SQLException类它简化了应用端的错误处理。可以使用java.sql.SQLException.getSQLState()方法获取错误码该方法会返回一个包含预定义ANSI SQLSTATE错误码的字符串
PreparedStatement ps;try {ps conn.prepareStatement(INSERT INTO Person(id, name, age) values (1, John, unparseableString));
} catch (SQLException e) {switch (e.getSQLState()) {case 0700B:System.out.println(Conversion failure);break;case 42000:System.out.println(Parsing error);break;default:System.out.println(Unprocessed error: e.getSQLState());break;}
}下表中列出了Ignite目前支持的所有ANSI SQLSTATE错误码未来这个列表可能还会扩展
代码描述0700B转换失败比如一个字符串表达式无法解析成数值或者日期0700E无效的事务隔离级别08001驱动接入集群失败08003连接意外地处于关闭状态08004连接被集群拒绝08006通信中发生I/O错误22004不允许的空值22023不支持的参数类型23000违反了数据完整性约束24000无效的结果集状态0A000不支持的操作40001并发更新冲突具体请参见并发更新章节的介绍。42000查询解析异常50000Ignite内部错误这个代码不是ANSI定义的属于Ignite特有的错误获取java.sql.SQLException的错误信息可以了解更多的细节
#9.JDBC客户端驱动
#9.1.JDBC客户端驱动
JDBC客户端节点模式驱动使用客户端节点连接接入集群这要求开发者提供一个完整的Spring XML配置作为JDBC连接串的一部分然后拷贝下面所有的jar文件到应用或者SQL工具的类路径中
{IGNITE_HOME}\libs目录下的所有jar文件{IGNITE_HOME}\ignite-indexing和{IGNITE_HOME}\ignite-spring目录下的所有jar文件
这个驱动很重而且可能不支持Ignite的最新SQL特性但是因为它底层使用客户端节点连接它可以执行分布式查询然后在应用端直接对结果进行汇总。
JDBC连接URL的规则如下
jdbc:ignite:cfg://[params]config_url其中
config_url是必需的表示指向Ignite客户端节点配置文件的任意合法URL当驱动试图建立到集群的连接时这个节点会在Ignite JDBC客户端节点驱动中启动params是可选的格式如下
param1value1:param2value2:...:paramNvalueN驱动类名为org.apache.ignite.IgniteJdbcDriver比如下面的代码展示了如何打开一个到集群的JDBC连接
// Register JDBC driver.
Class.forName(org.apache.ignite.IgniteJdbcDriver);// Open JDBC connection (cache name is not specified, which means that we use default cache).
Connection conn DriverManager.getConnection(jdbc:ignite:cfg://file:///etc/config/ignite-jdbc.xml);安全连接
关于如何保护JDBC客户端驱动的更多信息请参见高级安全的相关文档。
#9.1.1.支持的参数
属性描述默认值cache缓存名如果未定义会使用默认的缓存区分大小写nodeId要执行的查询所在节点的Id对于在本地查询是有用的local查询只在本地节点执行这个参数和nodeId参数都是通过指定节点来限制数据集falsecollocated优化标志当Ignite执行一个分布式查询时它会向单个的集群节点发送子查询如果提前知道要查询的数据已经被并置到同一个节点Ignite会有显著的性能提升和拓扑优化falsedistributedJoins可以在非并置的数据上使用分布式关联。falsestreaming通过INSERT语句为本链接开启批量数据加载模式具体可以参照后面的流模式相关章节。falsestreamingAllowOverwrite通知Ignite对于重复的已有键覆写它的值而不是忽略它们具体可以参照后面的流模式相关章节。falsestreamingFlushFrequency超时时间毫秒数据流处理器用于刷新数据数据默认会在连接关闭时刷新具体可以参照后面的流模式相关章节。0streamingPerNodeBufferSize数据流处理器的每节点缓冲区大小具体可以参照后面的流模式相关章节。1024streamingPerNodeParallelOperations数据流处理器的每节点并行操作数。具体可以参照后面的流模式相关章节。16transactionsAllowed目前已经支持了ACID事务但是仅仅在键-值API层面在SQL层面Ignite支持原子性还不支持事务一致性这意味着使用这个功能的时候驱动可能抛出Transactions are not supported这样的异常。但是有时需要使用事务语法即使不需要事务语义比如一些BI工具会一直强制事务行为也需要将该参数配置为true以避免异常。falsemultipleStatementsAllowedJDBC驱动可以同时处理多个SQL语句并且返回多个ResultSet对象如果该参数为false多个语句的查询会返回错误。falselazy查询延迟执行。Ignite默认会将所有的结果集放入内存然后将其返回给客户端对于不太大的结果集这样会提供较好的性能并且使内部的数据库锁时间最小化因此提高了并发能力。但是如果相对于可用内存来说结果集过大那么会导致频繁的GC暂停甚至OutOfMemoryError如果使用这个标志可以提示Ignite延迟加载结果集这样可以在不大幅降低性能的前提下最大限度地减少内存的消耗。falseskipReducerOnUpdate开启服务端的更新特性。当Ignite执行DML操作时首先它会获取所有受影响的中间行给查询发起方进行分析通常被称为汇总然后会准备一个更新值的批量发给远程节点。这个方式可能影响性能如果一个DML操作会移动大量数据条目时还可能会造成网络堵塞。使用这个标志可以提示Ignite在对应的远程节点上进行中间行的分析和更新。默认值为false这意味着会首先获取中间行然后发给查询发起方。false
#9.1.2.流模式
使用JDBC驱动可以以流模式批处理模式将数据注入Ignite集群。这时驱动会在内部实例化IgniteDataStreamer然后将数据传给它。要激活这个模式可以在JDBC连接串中增加streaming参数并且设置为true
// Register JDBC driver.
Class.forName(org.apache.ignite.IgniteJdbcDriver);// Opening connection in the streaming mode.
Connection conn DriverManager.getConnection(jdbc:ignite:cfg://streamingtruefile:///etc/config/ignite-jdbc.xml);目前流模式只支持INSERT操作对于想更快地将数据预加载进缓存的场景非常有用。JDBC驱动定义了多个连接参数来影响流模式的行为这些参数已经在上述的参数表中列出。
缓存名
确保在JDBC连接字符串中通过cache参数为流操作指定目标缓存。如果未指定缓存或缓存与流式DML语句中使用的表不匹配则更新会被忽略。
这些参数几乎覆盖了IgniteDataStreamer的所有常规配置这样就可以根据需要更好地调整流处理器。关于如何配置流处理器可以参考流处理器的相关文档来了解更多的信息。
基于时间的刷新
默认情况下当要么连接关闭要么达到了streamingPerNodeBufferSize数据才会被刷新如果希望按照时间的方式来刷新那么可以调整streamingFlushFrequency参数。
// Register JDBC driver.
Class.forName(org.apache.ignite.IgniteJdbcDriver);// Opening a connection in the streaming mode and time based flushing set.
Connection conn DriverManager.getConnection(jdbc:ignite:cfg://streamingtrue:streamingFlushFrequency1000file:///etc/config/ignite-jdbc.xml);PreparedStatement stmt conn.prepareStatement(INSERT INTO Person(_key, name, age) VALUES(CAST(? as BIGINT), ?, ?));// Adding the data.
for (int i 1; i 100000; i) {// Inserting a Person object with a Long key.stmt.setInt(1, i);stmt.setString(2, John Smith);stmt.setInt(3, 25);stmt.execute();
}conn.close();// Beyond this point, all data is guaranteed to be flushed into the cache.#9.2.示例
要处理集群中的数据需要使用下面的一种方式来创建一个JDBCConnection对象
// Register JDBC driver.
Class.forName(org.apache.ignite.IgniteJdbcDriver);// Open JDBC connection (cache name is not specified, which means that we use default cache).
Connection conn DriverManager.getConnection(jdbc:ignite:cfg://file:///etc/config/ignite-jdbc.xml);之后就可以执行SELECTSQL查询了
// Query names of all people.
ResultSet rs conn.createStatement().executeQuery(select name from Person);while (rs.next()) {String name rs.getString(1);
}// Query people with specific age using prepared statement.
PreparedStatement stmt conn.prepareStatement(select name, age from Person where age ?);stmt.setInt(1, 30);ResultSet rs stmt.executeQuery();while (rs.next()) {String name rs.getString(name);int age rs.getInt(age);
}此外可以使用DML语句对数据进行修改。
#9.2.1.INSERT
// Insert a Person with a Long key.
PreparedStatement stmt conn.prepareStatement(INSERT INTO Person(_key, name, age) VALUES(CAST(? as BIGINT), ?, ?));stmt.setInt(1, 1);
stmt.setString(2, John Smith);
stmt.setInt(3, 25);stmt.execute();#9.2.2.MERGE
// Merge a Person with a Long key.
PreparedStatement stmt conn.prepareStatement(MERGE INTO Person(_key, name, age) VALUES(CAST(? as BIGINT), ?, ?));stmt.setInt(1, 1);
stmt.setString(2, John Smith);
stmt.setInt(3, 25);stmt.executeUpdate();#9.2.3.UPDATE
// Update a Person.
conn.createStatement().executeUpdate(UPDATE Person SET age age 1 WHERE age 25);#9.2.4.DELETE
conn.createStatement().execute(DELETE FROM Person WHERE age 25);#10.ODBC驱动
#10.1.ODBC驱动
#10.1.1.概述
Ignite包括一个ODBC驱动可以通过标准SQL查询和原生ODBC API查询和修改存储于分布式缓存中的数据。
要了解ODBC的细节可以参照ODBC开发者参考。
Ignite的ODBC驱动实现了ODBC API的3.0版。
#10.1.2.集群配置
Ignite的ODBC驱动在Windows中被视为一个动态库在Linux中被视为一个共享对象应用不会直接加载它而是在必要时使用一个驱动加载器API来加载和卸载ODBC驱动。
Ignite的ODBC驱动在内部使用TCP来接入Ignite集群集群范围的连接参数可以通过IgniteConfiguration.clientConnectorConfiguration属性来配置
XMLJava
IgniteConfiguration cfg new IgniteConfiguration();ClientConnectorConfiguration clientConnectorCfg new ClientConnectorConfiguration();
cfg.setClientConnectorConfiguration(clientConnectorCfg);客户端连接器配置支持下面的参数
属性描述默认值host绑定的主机名或者IP地址如果为null会绑定localhostnullport绑定的TCP端口如果指定的端口被占用Ignite会使用portRange属性寻找其它的可用端口。10800portRange定义尝试绑定的端口范围。比如port配置为10800并且portRange为100那么服务端会按照顺序去尝试绑定[10800, 10900]范围内的端口直到找到可用的端口。100maxOpenCursorsPerConnection单个连接可以同时打开的最大游标数。128threadPoolSize线程池中负责请求处理的线程数。MAX(8, CPU核数)socketSendBufferSizeTCP套接字发送缓冲区大小如果配置为0会使用系统默认值0socketReceiveBufferSizeTCP套接字接收缓冲区大小如果配置为0会使用系统默认值。0tcpNoDelay是否使用TCP_NODELAY选项。trueidleTimeout客户端连接的空闲超时时间。如果空闲时间超过配置的超时时间客户端会自动断开与服务端的连接。如果该参数配置为0或者为负值空闲超时会被禁用。0isOdbcEnabled是否允许通过ODBC访问。trueisThinClientEnabled是否允许通过瘦客户端访问。true
可以通过如下方式修改参数
XMLJava
IgniteConfiguration cfg new IgniteConfiguration();
...
ClientConnectorConfiguration clientConnectorCfg new ClientConnectorConfiguration();clientConnectorCfg.setHost(127.0.0.1);
clientConnectorCfg.setPort(12345);
clientConnectorCfg.setPortRange(2);
clientConnectorCfg.setMaxOpenCursorsPerConnection(512);
clientConnectorCfg.setSocketSendBufferSize(65536);
clientConnectorCfg.setSocketReceiveBufferSize(131072);
clientConnectorCfg.setThreadPoolSize(4);cfg.setClientConnectorConfiguration(clientConnectorCfg);
...通过ClientListenerProcessor从ODBC驱动端建立的连接也是可以配置的关于如何从驱动端修改连接的配置可以看这里。
#10.1.3.线程安全
Ignite ODBC驱动的当前实现仅在连接层提供了线程安全这意味着如果没有额外的同步处理多线程无法访问同一个连接。不过可以为每个线程创建独立的连接然后同时使用。
#10.1.4.环境要求
Ignite的ODBC驱动官方在如下环境中进行了测试
OSWindowsXP及以上32位和64位版本 Windows Server2008及以上32位和64位版本 Ubuntu14.x和15.x64位C编译器MS Visual C (10.0及以上), g (4.4.0及以上)Visual Studio2010及以上
#10.1.5.构建ODBC驱动
在Windows中Ignite提供了预构建的32位和64位驱动的安装器因此如果只是想在Windows中安装驱动那么直接看下面的安装驱动章节就可以了。
对于Linux环境安装之前还是需要进行构建因此如果使用的是Linux或者使用Windows但是仍然想自己构建驱动那么往下看。
Ignite的ODBC驱动的源代码随着Ignite版本一起发布在使用之前可以自行构建。
因为ODBC驱动是用C编写的因此它是作为Ignite C的一部分提供的并且依赖于一些C库具体点说依赖于utils和binaryIgnite库这就意味着在构建ODBC驱动本身之前需要先构建它们。
这里假定使用的是二进制版本如果使用的是源代码版本那么需要将所有使用的%IGNITE_HOME%\platforms\cpp替换为%IGNITE_HOME%\modules\platforms\cpp。
#10.1.5.1.在Windows上构建
如果要在Windows上构建ODBC驱动需要MS Visual Studio 2010及以后的版本。一旦打开了Ignite方案%IGNITE_HOME%\platforms\cpp\project\vs\ignite.sln(或者ignite_86.sln32位平台)在方案浏览器中点击odbc项目然后选择“Build”Visual Studio会自动地检测并且构建所有必要的依赖。
.sln文件的路径可能会有所不同具体取决于是从源文件还是从二进制文件进行构建。如果在%IGNITE_HOME%\platforms\cpp\project\vs\中找不到.sln文件可以尝试在%IGNITE_HOME%\modules\platforms\cpp\project\vs\中查找。
注意
如果使用VS 2015及以后的版本MSVC14.0及以后需要将legacy_stdio_definitions.lib作为额外的库加入odbc项目的链接器配置以构建项目。要在IDE中将库文件加入链接器可以打开项目节点的上下文菜单选择Properties然后在Project Properties对话框中选择Linker然后编辑Linker Input这时就可以将legacy_stdio_definitions.lib加入分号分割的列表中。
构建过程完成之后会生成ignite.odbc.dll文件对于64位版本位于%IGNITE_HOME%\platforms\cpp\project\vs\x64\Release中对于32位版本位于%IGNITE_HOME%\platforms\cpp\project\vs\Win32\Release中。
注意
确认为系统使用相应的驱动32位或64位。
#10.1.5.2.在Windows中构建安装器
为了简化安装构建完驱动之后可能想构建安装器Ignite使用WiX工具包来生成ODBC的安装器因此需要下载并安装WiX记得一定要把Wix工具包的bin目录加入PATH变量中。
一切就绪之后打开终端然后定位到%IGNITE_HOME%\platforms\cpp\odbc\install目录按顺序执行如下的命令来构建安装器
64位32位
candle.exe ignite-odbc-amd64.wxs
light.exe -ext WixUIExtension ignite-odbc-amd64.wixobj完成之后目录中会出现ignite-odbc-amd64.msi和ignite-odbc-x86.msi文件然后就可以使用它们进行安装了。
#10.1.5.3.在Linux上构建
在一个基于Linux的操作系统中如果要构建及使用Ignite ODBC驱动需要安装选择的ODBC驱动管理器Ignite ODBC驱动已经使用UnixODBC进行了测试。
环境要求
C编译器cmake 3.6JDKopenssl包括头文件unixODBC。
下面列出了几种流行发行版的安装说明
Ubuntu 18.04/20.04CentOS/RHEL 7CentOS/RHEL 8
sudo apt-get install -y build-essential cmake openjdk-11-jdk unixodbc-dev libssl-dev提示
JDK只用于构建过程并不会用于ODBC驱动。
构建ODBC驱动
为cmake创建一个构建目录将其称为${CPP_BUILD_DIR}可选选择安装目录前缀默认为/usr/local将其称为${CPP_INSTALL_DIR}通过如下命令构建和安装驱动
UbuntuCentOS/RHEL
cd ${CPP_BUILD_DIR}
cmake -DCMAKE_BUILD_TYPERelease -DWITH_ODBCON ${IGNITE_HOME}/platforms/cpp -DCMAKE_INSTALL_PREFIX${CPP_INSTALL_DIR}
make
sudo make install构建过程完成后可以通过如下命令找到ODBC驱动位于何处
whereis libignite-odbc路径很可能是/usr/local/lib/libignite-odbc.so。
#10.1.6.安装ODBC驱动
要使用ODBC驱动首先要在系统中进行注册因此ODBC驱动管理器必须能找到它。
#10.1.6.1.在Windows上安装
在32位的Windows上需要使用32位版本的驱动而在64位的Windows上可以使用64位和32位版本的驱动也可以在64位的Windows上同时安装32位和64位版本的驱动这样32位和64位的应用都可以使用驱动。
使用安装器进行安装
注意
首先要安装微软的Microsoft Visual C 2010 Redistributable 32位或者64位包。
这是最简单的方式也是建议的方式只需要启动指定版本的安装器即可
32位%IGNITE_HOME%\platforms\cpp\bin\odbc\ignite-odbc-x86.msi64位%IGNITE_HOME%\platforms\cpp\bin\odbc\ignite-odbc-amd64.msi
手动安装
要在Windows上手动安装ODBC驱动首先要为驱动在文件系统中选择一个目录选择一个位置后就可以把驱动放在哪并且确保所有的驱动依赖可以被解析也就是说它们要么位于%PATH%要么和驱动DLL位于同一个目录。
之后就需要使用%IGNITE_HOME%/platforms/cpp/odbc/install目录下的安装脚本之一注意执行这些脚本很可能需要管理员权限。
x86AMD64
install_x86 absolute_path_to_32_bit_driver#10.1.6.2.在Linux上安装
要在Linux上构建和安装ODBC驱动首先需要安装ODBC驱动管理器Ignite ODBC驱动已经和UnixODBC进行了测试。
如果已经构建完成并且执行了make install命令libignite-odbc.so很可能会位于/usr/local/lib要在ODBC驱动管理器中安装ODBC驱动并且可以使用需要按照如下的步骤进行操作
确保链接器可以定位ODBC驱动的所有依赖。可以使用ldd命令像如下这样进行检查假定ODBC驱动位于/usr/local/lib:ldd /usr/local/lib/libignite-odbc.so如果存在到其它库的无法解析的链接需要将这些库文件所在的目录添加到LD_LIBRARY_PATH编辑$IGNITE_HOME/platforms/cpp/odbc/install/ignite-odbc-install.ini文件并且确保Apache Ignite段的Driver参数指向libignite-odbc.so所在的位置要安装Ignite的ODBC驱动可以使用如下的命令odbcinst -i -d -f $IGNITE_HOME/platforms/cpp/odbc/install/ignite-odbc-install.ini要执行这条命令很可能需要root权限。
到现在为止Ignite的ODBC驱动已经安装好了并且可以用了可以像其它ODBC驱动一样连接、使用。
#10.2.连接串和DSN
#10.2.1.连接串格式
Ignite的ODBC驱动支持标准的连接串格式下面是正常的语法
connection-string :: empty-string[;] | attribute[;] | attribute; connection-string
empty-string ::
attribute :: attribute-keywordattribute-value | DRIVER[{]attribute-value[}]
attribute-keyword :: identifier
attribute-value :: character-string简单来说连接串就是一个字符串其中包含了用分号分割的参数。
#10.2.2.支持的参数
Ignite的ODBC驱动可以使用一些连接串/DSN参数所有的参数都是大小写不敏感的因此ADDRESSAddressaddress都是有效的参数名并且指向的是同一个参数。如果参数未指定会使用默认值其中的一个例外是ADDRESS属性如果未指定会使用SERVER和PORT属性代替
属性关键字描述默认值ADDRESS要连接的远程节点的地址格式为host[:port]。比如localhost, example.com:12345, 127.0.0.1, 192.168.3.80:5893如果指定了这个属性SERVER和PORT将会被忽略。SERVER要连接的节点地址如果指定了ADDRESS属性本属性会被忽略。PORT节点的OdbcProcessor监听的端口,如果指定了ADDRESS属性本属性会被忽略。10800USERSQL连接的用户名。如果服务端开启了认证该参数为必需。“”PASSWORDSQL连接的密码。如果服务端开启了认证该参数为必需。“”SCHEMA模式名。PUBLICDSN要连接的DSN名PAGE_SIZE数据源的响应中返回的行数默认值会适用于大多数场景小些的值会导致获取数据变慢大些的值会导致驱动的额外内存占用以及获取下一页时的额外延迟。1024DISTRIBUTED_JOINS为在ODBC连接上执行的所有查询开启非并置的分布式关联特性。falseENFORCE_JOIN_ORDER强制SQL查询中表关联顺序如果设置为true查询优化器在关联时就不会对表进行再排序。falsePROTOCOL_VERSION使用的ODBC协议版本目前支持如下的版本2.1.0、2.1.5、2.3.0、2.3.2和2.5.0因为向后兼容也可以使用协议的早期版本。2.3.0REPLICATED_ONLY配置查询只在全复制的表上执行这是个提示用于更高效地执行。falseCOLLOCATED如果SQL语句包含按主键或关联键对结果集进行分组的GROUP BY子句可以将此参数设置为true。当Ignite执行分布式查询时会向单个集群节点发送子查询如果事先知道待查询的数据是在同一个节点上并置在一起的并且是按主键或关联键分组的那么Ignite通过在参与查询的每个节点本地分组数据来实现显著的性能和网络优化。falseLAZY查询延迟执行。Ignite默认会将所有的结果集放入内存然后将其返回给客户端对于不太大的结果集这样会提供较好的性能并且使内部的数据库锁时间最小化因此提高了并发能力。但是如果相对于可用内存来说结果集过大那么会导致频繁的GC暂停甚至OutOfMemoryError如果使用这个标志可以提示Ignite延迟加载结果集这样可以在不大幅降低性能的前提下最大限度地减少内存的消耗。falseSKIP_REDUCER_ON_UPDATE开启服务端的更新特性。当Ignite执行DML操作时首先它会获取所有受影响的中间行给查询发起方进行分析通常被称为汇总然后会准备一个更新值的批量发给远程节点。这个方式可能影响性能如果一个DML操作会移动大量数据条目时还可能会造成网络堵塞。使用这个标志可以提示Ignite在对应的远程节点上进行中间行的分析和更新。默认值为false这意味着会首先获取中间行然后发给查询发起方。falseSSL_MODE确定服务端是否需要SSL连接。可以根据需要使用require或者disable。SSL_KEY_FILE指定包含服务端SSL私钥的文件名。SSL_CERT_FILE指定包含SSL服务器证书的文件名。SSL_CA_FILE指定包含SSL服务器证书颁发机构CA的文件名。
#10.2.3.连接串示例
下面的串可以用于SQLDriverConnectODBC调用来建立与Ignite节点的连接。
认证指定缓存默认缓存DSN自定义页面大小
DRIVER{Apache Ignite};
ADDRESSlocalhost:10800;
SCHEMAsomecachename;
USERyourusername;
PASSWORDyourpassword;
SSL_MODE[require|disable];
SSL_KEY_FILEpath_to_private_key;
SSL_CERT_FILEpath_to_client_certificate;
SSL_CA_FILEpath_to_trusted_certificates#10.2.4.配置DSN
如果要使用DSN(数据源名)来进行连接可以使用同样的参数。
要在Windows上配置DSN需要使用一个叫做odbcad3232位x86系统/odbcad6464位的系统工具这是一个ODBC数据源管理器。
安装DSN工具时如果使用的是预构建的msi文件一定要先安装Microsoft Visual C 201032位x86或者64位x64。
要启动这个工具打开Control panel-Administrative Tools-数据源ODBC当ODBC数据源管理器启动后选择Add...-Apache Ignite然后以正确的方式配置DSN。 在Linux上配置DSN需要找到odbc.ini文件这个文件的位置各个发行版有所不同依赖于发行版使用的驱动管理器比如如果使用unixODBC那么可以执行如下的命令来输出系统级的ODBC相关信息
odbcinst -j使用SYSTEM DATA SOURCES和USER DATA SOURCES属性可以定位odbc.ini文件。
找到odbc.ini文件之后可以用任意编辑器打开它然后像下面这样添加DSN片段
[DSN Name]
descriptionInsert your description here
driverApache Ignite
Other arguments here...#10.3.查询和修改数据
#10.3.1.概述
本章会详细描述如何接入Ignite集群如何使用ODBC驱动执行各种SQL查询。
在实现层Ignite的ODBC驱动使用SQL字段查询来获取Ignite缓存中的数据这意味着通过ODBC只可以访问这些集群配置中定义的字段。
另外ODBC驱动支持DML这意味着通过ODBC连接不仅仅可以读取数据还可以修改数据。
提示
这里是完整的ODBC示例。
#10.3.2.配置Ignite集群
第一步需要对集群节点进行配置这个配置需要包含缓存的配置以及定义了QueryEntities的属性。如果应用当前场景是ODBC驱动要通过SQL语句进行数据的查询和修改QueryEntities是必须的或者也可以使用DDL创建表。
DDLSpring XML
SQLHENV env;// Allocate an environment handle
SQLAllocHandle(SQL_HANDLE_ENV, SQL_NULL_HANDLE, env);// Use ODBC ver 3
SQLSetEnvAttr(env, SQL_ATTR_ODBC_VERSION, reinterpret_castvoid*(SQL_OV_ODBC3), 0);SQLHDBC dbc;// Allocate a connection handle
SQLAllocHandle(SQL_HANDLE_DBC, env, dbc);// Prepare the connection string
SQLCHAR connectStr[] DSNMy Ignite DSN;// Connecting to Ignite Cluster.
SQLDriverConnect(dbc, NULL, connectStr, SQL_NTS, NULL, 0, NULL, SQL_DRIVER_COMPLETE);SQLHSTMT stmt;// Allocate a statement handle
SQLAllocHandle(SQL_HANDLE_STMT, dbc, stmt);SQLCHAR query1[] CREATE TABLE Person ( id LONG PRIMARY KEY, firstName VARCHAR, lastName VARCHAR, salary FLOAT) WITH \templatepartitioned\;SQLExecDirect(stmt, query1, SQL_NTS);SQLCHAR query2[] CREATE TABLE Organization ( id LONG PRIMARY KEY, name VARCHAR) WITH \templatepartitioned\;SQLExecDirect(stmt, query2, SQL_NTS);SQLCHAR query3[] CREATE INDEX idx_organization_name ON Organization (name);SQLExecDirect(stmt, query3, SQL_NTS);从上述配置中可以看出定义了两个缓存包含了Person和Organization类型的数据它们都列出了使用SQL可以读写的字段和索引。
#10.3.3.接入集群
配置好然后启动集群就可以从ODBC驱动端接入了。如何做呢准备一个有效的连接串然后连接时将其作为一个参数传递给ODBC驱动就可以了。
另外也可以像下面这样使用一个预定义的DSN来接入。
SQLHENV env;// Allocate an environment handle
SQLAllocHandle(SQL_HANDLE_ENV, SQL_NULL_HANDLE, env);// Use ODBC ver 3
SQLSetEnvAttr(env, SQL_ATTR_ODBC_VERSION, reinterpret_castvoid*(SQL_OV_ODBC3), 0);SQLHDBC dbc;// Allocate a connection handle
SQLAllocHandle(SQL_HANDLE_DBC, env, dbc);// Prepare the connection string
SQLCHAR connectStr[] DSNMy Ignite DSN;// Connecting to Ignite Cluster.
SQLRETURN ret SQLDriverConnect(dbc, NULL, connectStr, SQL_NTS, NULL, 0, NULL, SQL_DRIVER_COMPLETE);if (!SQL_SUCCEEDED(ret))
{SQLCHAR sqlstate[7] { 0 };SQLINTEGER nativeCode;SQLCHAR errMsg[BUFFER_SIZE] { 0 };SQLSMALLINT errMsgLen static_castSQLSMALLINT(sizeof(errMsg));SQLGetDiagRec(SQL_HANDLE_DBC, dbc, 1, sqlstate, nativeCode, errMsg, errMsgLen, errMsgLen);std::cerr Failed to connect to Apache Ignite: reinterpret_castchar*(sqlstate) : reinterpret_castchar*(errMsg) , Native error code: nativeCode std::endl;// Releasing allocated handles.SQLFreeHandle(SQL_HANDLE_DBC, dbc);SQLFreeHandle(SQL_HANDLE_ENV, env);return;
}#10.3.4.查询数据
都准备好后就可以使用ODBC API执行SQL查询了。
SQLHSTMT stmt;// Allocate a statement handle
SQLAllocHandle(SQL_HANDLE_STMT, dbc, stmt);SQLCHAR query[] SELECT firstName, lastName, salary, Organization.name FROM Person INNER JOIN \Organization\.Organization ON Person.orgId Organization.id;
SQLSMALLINT queryLen static_castSQLSMALLINT(sizeof(queryLen));SQLRETURN ret SQLExecDirect(stmt, query, queryLen);if (!SQL_SUCCEEDED(ret))
{SQLCHAR sqlstate[7] { 0 };SQLINTEGER nativeCode;SQLCHAR errMsg[BUFFER_SIZE] { 0 };SQLSMALLINT errMsgLen static_castSQLSMALLINT(sizeof(errMsg));SQLGetDiagRec(SQL_HANDLE_DBC, dbc, 1, sqlstate, nativeCode, errMsg, errMsgLen, errMsgLen);std::cerr Failed to perfrom SQL query upon Apache Ignite: reinterpret_castchar*(sqlstate) : reinterpret_castchar*(errMsg) , Native error code: nativeCode std::endl;
}
else
{// Printing the result set.struct OdbcStringBuffer{SQLCHAR buffer[BUFFER_SIZE];SQLLEN resLen;};// Getting a number of columns in the result set.SQLSMALLINT columnsCnt 0;SQLNumResultCols(stmt, columnsCnt);// Allocating buffers for columns.std::vectorOdbcStringBuffer columns(columnsCnt);// Binding colums. For simplicity we are going to use only// string buffers here.for (SQLSMALLINT i 0; i columnsCnt; i)SQLBindCol(stmt, i 1, SQL_C_CHAR, columns[i].buffer, BUFFER_SIZE, columns[i].resLen);// Fetching and printing data in a loop.ret SQLFetch(stmt);while (SQL_SUCCEEDED(ret)){for (size_t i 0; i columns.size(); i)std::cout std::setw(16) std::left columns[i].buffer ;std::cout std::endl;ret SQLFetch(stmt);}
}// Releasing statement handle.
SQLFreeHandle(SQL_HANDLE_STMT, stmt);列绑定
在上例中所有的列都绑定到SQL_C_CHAR这意味着获取时所有的值都会被转换成字符串这样做是为了简化获取时进行值转换是非常慢的因此默认的做法应该是与存储采用同样的方式进行获取。
#10.3.5.插入数据
要将新的数据插入集群ODBC端可以使用INSERT语句。
SQLHSTMT stmt;// Allocate a statement handle
SQLAllocHandle(SQL_HANDLE_STMT, dbc, stmt);SQLCHAR query[] INSERT INTO Person (id, orgId, firstName, lastName, resume, salary) VALUES (?, ?, ?, ?, ?, ?);SQLPrepare(stmt, query, static_castSQLSMALLINT(sizeof(query)));// Binding columns.
int64_t key 0;
int64_t orgId 0;
char name[1024] { 0 };
SQLLEN nameLen SQL_NTS;
double salary 0.0;SQLBindParameter(stmt, 1, SQL_PARAM_INPUT, SQL_C_SLONG, SQL_BIGINT, 0, 0, key, 0, 0);
SQLBindParameter(stmt, 2, SQL_PARAM_INPUT, SQL_C_SLONG, SQL_BIGINT, 0, 0, orgId, 0, 0);
SQLBindParameter(stmt, 3, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, sizeof(name), sizeof(name), name, 0, nameLen);
SQLBindParameter(stmt, 4, SQL_PARAM_INPUT, SQL_C_DOUBLE, SQL_DOUBLE, 0, 0, salary, 0, 0);// Filling cache.
key 1;
orgId 1;
strncpy(name, John, sizeof(name));
salary 2200.0;SQLExecute(stmt);
SQLMoreResults(stmt);key;
orgId 1;
strncpy(name, Jane, sizeof(name));
salary 1300.0;SQLExecute(stmt);
SQLMoreResults(stmt);key;
orgId 2;
strncpy(name, Richard, sizeof(name));
salary 900.0;SQLExecute(stmt);
SQLMoreResults(stmt);key;
orgId 2;
strncpy(name, Mary, sizeof(name));
salary 2400.0;SQLExecute(stmt);// Releasing statement handle.
SQLFreeHandle(SQL_HANDLE_STMT, stmt);下面是不使用预编译语句插入Organization数据
SQLHSTMT stmt;// Allocate a statement handle
SQLAllocHandle(SQL_HANDLE_STMT, dbc, stmt);SQLCHAR query1[] INSERT INTO \Organization\.Organization (id, name)VALUES (1L, Some company);SQLExecDirect(stmt, query1, static_castSQLSMALLINT(sizeof(query1)));SQLFreeStmt(stmt, SQL_CLOSE);SQLCHAR query2[] INSERT INTO \Organization\.Organization (id, name)VALUES (2L, Some other company);SQLExecDirect(stmt, query2, static_castSQLSMALLINT(sizeof(query2)));// Releasing statement handle.
SQLFreeHandle(SQL_HANDLE_STMT, stmt);错误检查
为了简化上面的代码没有进行错误检查但是在生产环境中不要这样做。
#10.3.6.更新数据
下面使用UPDATE语句更新存储在集群中的部分人员的工资信息
void AdjustSalary(SQLHDBC dbc, int64_t key, double salary)
{SQLHSTMT stmt;// Allocate a statement handleSQLAllocHandle(SQL_HANDLE_STMT, dbc, stmt);SQLCHAR query[] UPDATE Person SET salary? WHERE id?;SQLBindParameter(stmt, 1, SQL_PARAM_INPUT,SQL_C_DOUBLE, SQL_DOUBLE, 0, 0, salary, 0, 0);SQLBindParameter(stmt, 2, SQL_PARAM_INPUT, SQL_C_SLONG,SQL_BIGINT, 0, 0, key, 0, 0);SQLExecDirect(stmt, query, static_castSQLSMALLINT(sizeof(query)));// Releasing statement handle.SQLFreeHandle(SQL_HANDLE_STMT, stmt);
}...
AdjustSalary(dbc, 3, 1200.0);
AdjustSalary(dbc, 1, 2500.0);#10.3.7.删除数据
最后使用DELETE语句删除部分记录
void DeletePerson(SQLHDBC dbc, int64_t key)
{SQLHSTMT stmt;// Allocate a statement handleSQLAllocHandle(SQL_HANDLE_STMT, dbc, stmt);SQLCHAR query[] DELETE FROM Person WHERE id?;SQLBindParameter(stmt, 1, SQL_PARAM_INPUT, SQL_C_SLONG, SQL_BIGINT,0, 0, key, 0, 0);SQLExecDirect(stmt, query, static_castSQLSMALLINT(sizeof(query)));// Releasing statement handle.SQLFreeHandle(SQL_HANDLE_STMT, stmt);
}...
DeletePerson(dbc, 1);
DeletePerson(dbc, 4);#10.3.8.通过参数数组进行批处理
Ignite的ODBC驱动支持在DML语句中通过参数数组进行批处理。
还是使用上述插入数据的示例但是只调用一次SQLExecute:
SQLHSTMT stmt;// Allocating a statement handle.
SQLAllocHandle(SQL_HANDLE_STMT, dbc, stmt);SQLCHAR query[] INSERT INTO Person (id, orgId, firstName, lastName, resume, salary) VALUES (?, ?, ?, ?, ?, ?);SQLPrepare(stmt, query, static_castSQLSMALLINT(sizeof(query)));// Binding columns.
int64_t key[4] {0};
int64_t orgId[4] {0};
char name[1024 * 4] {0};
SQLLEN nameLen[4] {0};
double salary[4] {0};SQLBindParameter(stmt, 1, SQL_PARAM_INPUT, SQL_C_SLONG, SQL_BIGINT, 0, 0, key, 0, 0);
SQLBindParameter(stmt, 2, SQL_PARAM_INPUT, SQL_C_SLONG, SQL_BIGINT, 0, 0, orgId, 0, 0);
SQLBindParameter(stmt, 3, SQL_PARAM_INPUT, SQL_C_CHAR, SQL_VARCHAR, 1024, 1024, name, 0, nameLen);
SQLBindParameter(stmt, 4, SQL_PARAM_INPUT, SQL_C_DOUBLE, SQL_DOUBLE, 0, 0, salary, 0, 0);// Filling cache.
key[0] 1;
orgId[0] 1;
strncpy(name, John, 1023);
salary[0] 2200.0;
nameLen[0] SQL_NTS;key[1] 2;
orgId[1] 1;
strncpy(name 1024, Jane, 1023);
salary[1] 1300.0;
nameLen[1] SQL_NTS;key[2] 3;
orgId[2] 2;
strncpy(name 1024 * 2, Richard, 1023);
salary[2] 900.0;
nameLen[2] SQL_NTS;key[3] 4;
orgId[3] 2;
strncpy(name 1024 * 3, Mary, 1023);
salary[3] 2400.0;
nameLen[3] SQL_NTS;// Asking the driver to store the total number of processed argument sets
// in the following variable.
SQLULEN setsProcessed 0;
SQLSetStmtAttr(stmt, SQL_ATTR_PARAMS_PROCESSED_PTR, setsProcessed, SQL_IS_POINTER);// Setting the size of the arguments array. This is 4 in our case.
SQLSetStmtAttr(stmt, SQL_ATTR_PARAMSET_SIZE, reinterpret_castSQLPOINTER(4), 0);// Executing the statement.
SQLExecute(stmt);// Releasing the statement handle.
SQLFreeHandle(SQL_HANDLE_STMT, stmt);注意
注意这种类型的批处理目前只支持INSERT、UPDATE、 DELETE、和MERGE语句还不支持SELECTdata-at-execution功能也不支持通过参数数组进行批处理。
#10.3.9.流处理
Ignite的ODBC驱动可以通过SET STREAMING命令对流化数据进行批量处理具体可以看SET STREAMING的相关内容。
注意
流处理模式中参数数组和data-at-execution参数是不支持的。
#10.4.规范
#10.4.1.概述
ODBC定义了若干接口一致性级别在本章中可以知道Ignite的ODBC驱动支持了哪些特性。
#10.4.2.核心接口一致性
特性支持程度备注通过调用SQLAllocHandle和SQLFreeHandle来分配和释放所有处理器类型是使用SQLFreeStmt函数的所有形式是通过调用SQLBindCol绑定列结果集是通过调用SQLBindParameter和SQLNumParams处理动态参数包括参数数组只针对输入方向是指定绑定偏移量是使用数据执行对话框涉及SQLParamData和SQLPutData的调用是管理游标和游标名部分实现了SQLCloseCursorIgnite不支持命名游标通过调用SQLColAttributeSQLDescribeColSQLNumResultCols和SQLRowCount访问结果集的描述元数据是通过调用目录函数SQLColumnsSQLGetTypeInfoSQLStatistics和SQLStatistics查询数据字典部分不支持SQLStatistics通过调用SQLConnectSQLDataSourcesSQLDisconnect和SQLDriverConnect管理数据源和连接通过SQLDrivers获取驱动的信息不管支持ODBC那个级别。是通过调用SQLExecDirectSQLExecute和SQLPrepare预编译和执行SQL语句。是通过调用SQLFetch或者将FetchOrientation参数设置为SQL_FETCH_NEXT之后调用SQLFetchScroll获取一个结果集或者多行数据中的一行只能向前是通过调用SQLGetData获得一个未绑定的列是通过调用SQLGetConnectAttr、SQLGetEnvAttr、SQLGetStmtAttr获取所有属性的当前值或者通过调用SQLSetConnectAttr、SQLSetEnvAttr、SQLSetStmtAttr将所有属性赋为默认值以及为某个属性赋为非默认值。部分并不支持所有属性通过调用SQLCopyDesc、SQLGetDescField、SQLGetDescRec、SQLSetDescField、SQLSetDescRec操作描述符的某字段。否通过调用SQLGetDiagField、SQLGetDiagRec获得诊断信息。是通过调用SQLGetFunctions和SQLGetInfo检测驱动兼容性以及通过调用SQLNativeSql在发送到数据源之前检测SQL语句中的任何文本代换的结果是使用SQLEndTran的语法提交一个事务驱动的核心级别不需要支持真事务因此应用无法指定SQL_ROLLBACK或者为SQL_ATTR_AUTOCOMMIT连接属性指定SQL_AUTOCOMMIT_OFF是调用SQLCancel取消数据执行对话框以及多线程环境中在另一个线程中取消ODBC函数的执行核心级别的接口一致性不需要支持函数的异步执行也不需要使用SQLCancel取消一个ODBC函数的异步执行。平台和ODBC驱动都不需要多线程地同时自主活动不过在多线程环境中ODBC驱动必须是线程安全的从应用来的请求的序列化是实现这个规范的一致的方式即使它导致了一系列的性能问题。否当前的ODBC驱动实现不支持异步执行通过调用SQLSpecialColumns获得表的行标识符SQL_BEST_ROWID。部分当前的实现总是返回空
#10.4.3.Level1接口一致性
特性支持程度备注指定数据库表和视图的模式使用两部分命名。是ODBC函数调用的真正异步执行在给定的连接上适用的函数要么是全同步的要么是全异步的。否使用可滚动的游标调用SQLFetchScroll时使用FetchOrientation参数而不是SQL_FETCH_NEXT可以在方法内访问结果集而不是只能向前。否通过调用SQLPrimaryKeys获得表的主键。部分目前返回空结果集。使用存储过程通过调用SQLProcedureColumns和SQLProcedures使用ODBC的转义序列进行存储过程数据字典的查询以及存储过程的调用。否通过调用SQLBrowseConnect通过交互式浏览可用的服务器接入一个数据源。否使用ODBC函数而不是SQL语句来执行某个数据库操作带有SQL_POSITION和SQL_REFRESH的SQLSetPos。否通过调用SQLMoreResults访问由批处理和存储过程生成的多结果集的内容。是划定跨越多个ODBC函数的事务边界获得真正的原子性以及在SQLEndTran中指定SQL_ROLLBACK的能力。否Ignite SQL不支持事务
#10.4.4.Level2接口一致性
特性支持程度备注使用三部分命名的数据库表和视图。否Ignite SQL不支持catalog。通过调用SQLDescribeParam描述动态参数。是不仅仅使用输入参数还使用输出参数以及输入/输出参数还有存储过程的结果。否Ignite SQL不支持输出参数。使用书签通过在第0列上调用SQLDescribeCol和SQLColAttribute获得书签通过调用SQLFetchScroll时将参数FetchOrientation配置为SQL_FETCH_BOOKMARK在书签上进行获取通过调用SQLBulkOperations时将参数配置为SQL_UPDATE_BY_BOOKMARK、SQL_DELETE_BY_BOOKMARK、SQL_FETCH_BY_BOOKMARK可以进行书签的更新、删除和获取操作。否Ignite SQL不支持书签。通过调用SQLColumnPrivileges、SQLForeignKeys、SQLTablePrivileges获取数据字典的高级信息。部分SQLForeignKeys已经实现但是返回空的结果集。通过在SQLBulkOperations中使用SQL_ADD或者在SQLSetPos中使用SQL_DELETE或SQL_UPDATE使用ODBC函数而不是SQL语句执行额外的数据库操作。否为某个语句开启ODBC函数的异步执行。否通过调用SQLSpecialColumns获得表的SQL_ROWVER列标识符。部分已实现但是返回空结果集。为SQL_ATTR_CONCURRENCY语句参数配置除了SQL_CONCUR_READ_ONLY以外的至少一个值。否登录请求以及SQL查询的超时功能(SQL_ATTR_LOGIN_TIMEOUT和SQL_ATTR_QUERY_TIMEOUT)。部分SQL_ATTR_QUERY_TIMEOUT支持已实现SQL_ATTR_LOGIN_TIMEOUT还未实现。修改默认隔离级别的功能在隔离级别为序列化时支持事务的功能。否Ignite SQL不支持事务。
#10.4.5.函数支持
函数名支持程度一致性级别SQLAllocHandle是CoreSQLBindCol是CoreSQLBindParameter是CoreSQLBrowseConnect否Level1SQLBulkOperations否Level1SQLCancel否CoreSQLCloseCursor是CoreSQLColAttribute是CoreSQLColumnPrivileges否Level2SQLColumns是CoreSQLConnect是CoreSQLCopyDesc否CoreSQLDataSourcesN/ACoreSQLDescribeCol是CoreSQLDescribeParam是Level2SQLDisconnect是CoreSQLDriverConnect是CoreSQLDriversN/ACoreSQLEndTran部分CoreSQLExecDirect是CoreSQLExecute是CoreSQLFetch是CoreSQLFetchScroll是CoreSQLForeignKeys部分Level2SQLFreeHandle是CoreSQLFreeStmt是CoreSQLGetConnectAttr部分CoreSQLGetCursorName否CoreSQLGetData是CoreSQLGetDescField否CoreSQLGetDescRec否CoreSQLGetDiagField是CoreSQLGetDiagRec是CoreSQLGetEnvAttr部分CoreSQLGetFunctions否CoreSQLGetInfo是CoreSQLGetStmtAttr部分CoreSQLGetTypeInfo是CoreSQLMoreResults是Level1SQLNativeSql是CoreSQLNumParams是CoreSQLNumResultCols是CoreSQLParamData是CoreSQLPrepare是CoreSQLPrimaryKeys部分Level1SQLProcedureColumns否Level1SQLProcedures否Level1SQLPutData是CoreSQLRowCount是CoreSQLSetConnectAttr部分CoreSQLSetCursorName否CoreSQLSetDescField否CoreSQLSetDescRec否CoreSQLSetEnvAttr部分CoreSQLSetPos否Level1SQLSetStmtAttr部分CoreSQLSpecialColumns部分CoreSQLStatistics否CoreSQLTablePrivileges否Level2SQLTables是Core
#10.4.6.环境属性一致性
特性支持程度一致性级别SQL_ATTR_CONNECTION_POOLING否可选SQL_ATTR_CP_MATCH否可选SQL_ATTR_ODBC_VER是CoreSQL_ATTR_OUTPUT_NTS是可选
#10.4.7.连接属性一致性
特性支持程度一致性级别SQL_ATTR_ACCESS_MODE否CoreSQL_ATTR_ASYNC_ENABLE否Level1/Level2SQL_ATTR_AUTO_IPD否Level2SQL_ATTR_AUTOCOMMIT否Level1SQL_ATTR_CONNECTION_DEAD是Level1SQL_ATTR_CONNECTION_TIMEOUT是Level2SQL_ATTR_CURRENT_CATALOG否Level2SQL_ATTR_LOGIN_TIMEOUT否Level2SQL_ATTR_ODBC_CURSORS否CoreSQL_ATTR_PACKET_SIZE否Level2否SQL_ATTR_QUIET_MODE否CoreSQL否_ATTR_TRACE否CoreSQL_AT否TR_TRACEFILE否CoreSQL_AT否TR_TRANSLATE_LIB否CoreSQL_ATTR_TRANSLATE_OPTION否CoreSQL_ATTR_TXN_ISOLATION否Level1/Level2
#10.4.8.语句属性一致性
特性支持程度一致性级别SQL_ATTR_APP_PARAM_DESC部分CoreSQL_ATTR_APP_ROW_DESC部分CoreSQL_ATTR_ASYNC_ENABLE否Level1/Level2SQL_ATTR_CONCURRENCY否Level1/Level2SQL_ATTR_CURSOR_SCROLLABLE否Level1SQL_ATTR_CURSOR_SENSITIVITY否Level2SQL_ATTR_CURSOR_TYPE否Level1/Level2SQL_ATTR_ENABLE_AUTO_IPD否Level2SQL_ATTR_FETCH_BOOKMARK_PTR否Level2SQL_ATTR_IMP_PARAM_DESC部分CoreSQL_ATTR_IMP_ROW_DESC部分CoreSQL_ATTR_KEYSET_SIZE否Level2SQL_ATTR_MAX_LENGTH否Level1SQL_ATTR_MAX_ROWS否Level1SQL_ATTR_METADATA_ID否CoreSQL_ATTR_NOSCAN否CoreSQL_ATTR_PARAM_BIND_OFFSET_PTR是CoreSQL_ATTR_PARAM_BIND_TYPE否CoreSQL_ATTR_PARAM_OPERATION_PTR否CoreSQL_ATTR_PARAM_STATUS_PTR是CoreSQL_ATTR_PARAMS_PROCESSED_PTR是CoreSQL_ATTR_PARAMSET_SIZE是CoreSQL_ATTR_QUERY_TIMEOUT是Level2SQL_ATTR_RETRIEVE_DATA否Level1SQL_ATTR_ROW_ARRAY_SIZE是CoreSQL_ATTR_ROW_BIND_OFFSET_PTR是CoreSQL_ATTR_ROW_BIND_TYPE是CoreSQL_ATTR_ROW_NUMBER否Level1SQL_ATTR_ROW_OPERATION_PTR否Level1SQL_ATTR_ROW_STATUS_PTR是CoreSQL_ATTR_ROWS_FETCHED_PTR是CoreSQL_ATTR_SIMULATE_CURSOR否Level2SQL_ATTR_USE_BOOKMARKS否Level2
#10.4.9.描述符头字段一致性
特性支持程度一致性级别SQL_DESC_ALLOC_TYPE否CoreSQL_DESC_ARRAY_SIZE否CoreSQL_DESC_ARRAY_STATUS_PTR否Core/Level1SQL_DESC_BIND_OFFSET_PTR否CoreSQL_DESC_BIND_TYPE否CoreSQL_DESC_COUNT否CoreSQL_DESC_ROWS_PROCESSED_PTR否Core
#10.4.10.描述符记录字段一致性
特性支持程度一致性级别SQL_DESC_AUTO_UNIQUE_VALUE否Level2SQL_DESC_BASE_COLUMN_NAME否CoreSQL_DESC_BASE_TABLE_NAME否Level1SQL_DESC_CASE_SENSITIVE否CoreSQL_DESC_CATALOG_NAME否Level2SQL_DESC_CONCISE_TYPE否CoreSQL_DESC_DATA_PTR否CoreSQL_DESC_DATETIME_INTERVAL_CODE否CoreSQL_DESC_DATETIME_INTERVAL_PRECISION否CoreSQL_DESC_DISPLAY_SIZE否CoreSQL_DESC_FIXED_PREC_SCALE否CoreSQL_DESC_INDICATOR_PTR否CoreSQL_DESC_LABEL否Level2SQL_DESC_LENGTH否CoreSQL_DESC_LITERAL_PREFIX否CoreSQL_DESC_LITERAL_SUFFIX否CoreSQL_DESC_LOCAL_TYPE_NAME否CoreSQL_DESC_NAME否CoreSQL_DESC_NULLABLE否CoreSQL_DESC_OCTET_LENGTH否CoreSQL_DESC_OCTET_LENGTH_PTR否CoreSQL_DESC_PARAMETER_TYPE否Core/Level2SQL_DESC_PRECISION否CoreSQL_DESC_ROWVER否Level1SQL_DESC_SCALE否CoreSQL_DESC_SCHEMA_NAME否Level1SQL_DESC_SEARCHABLE否CoreSQL_DESC_TABLE_NAME否Level1SQL_DESC_TYPE否CoreSQL_DESC_TYPE_NAME否CoreSQL_DESC_UNNAMED否CoreSQL_DESC_UNSIGNED否CoreSQL_DESC_UPDATABLE否Core
#10.4.11.SQL数据类型
下面是支持的SQL数据类型
数据类型是否支持SQL_CHAR是SQL_VARCHAR是SQL_LONGVARCHAR是SQL_WCHAR否SQL_WVARCHAR否SQL_WLONGVARCHAR否SQL_DECIMAL是SQL_NUMERIC否SQL_SMALLINT是SQL_INTEGER是SQL_REAL否SQL_FLOAT是SQL_DOUBLE是SQL_BIT是SQL_TINYINT是SQL_BIGINT是SQL_BINARY是SQL_VARBINARY是SQL_LONGVARBINARY是SQL_TYPE_DATE是SQL_TYPE_TIME是SQL_TYPE_TIMESTAMP是SQL_TYPE_UTCDATETIME否SQL_TYPE_UTCTIME否SQL_INTERVAL_MONTH否SQL_INTERVAL_YEAR否SQL_INTERVAL_YEAR_TO_MONTH否SQL_INTERVAL_DAY否SQL_INTERVAL_HOUR否SQL_INTERVAL_MINUTE否SQL_INTERVAL_SECOND否SQL_INTERVAL_DAY_TO_HOUR否SQL_INTERVAL_DAY_TO_MINUTE否SQL_INTERVAL_DAY_TO_SECOND否SQL_INTERVAL_HOUR_TO_MINUTE否SQL_INTERVAL_HOUR_TO_SECOND否SQL_INTERVAL_MINUTE_TO_SECOND否SQL_GUID是
#10.4.12.C数据类型
下面是支持的C数据类型
数据类型是否支持SQL_C_CHAR是SQL_C_WCHAR是SQL_C_SHORT是SQL_C_SSHORT是SQL_C_USHORT是SQL_C_LONG是SQL_C_SLONG是SQL_C_ULONG是SQL_C_FLOAT是SQL_C_DOUBLE是SQL_C_BIT是SQL_C_TINYINT是SQL_C_STINYINT是SQL_C_UTINYINT是SQL_C_BIGINT是SQL_C_SBIGINT是SQL_C_UBIGINT是SQL_C_BINARY是SQL_C_BOOKMARK否SQL_C_VARBOOKMARK否SQL_C_INTERVAL* (all interval types)否SQL_C_TYPE_DATE是SQL_C_TYPE_TIME是SQL_C_TYPE_TIMESTAMP是SQL_C_NUMERIC是SQL_C_GUID是
#10.5.数据类型
支持如下的SQL数据类型规范中列出
SQL_CHARSQL_VARCHARSQL_LONGVARCHARSQL_SMALLINTSQL_INTEGERSQL_FLOATSQL_DOUBLESQL_BITSQL_TINYINTSQL_BIGINTSQL_BINARYSQL_VARBINARYSQL_LONGVARBINARYSQL_GUIDSQL_DECIMALSQL_TYPE_DATESQL_TYPE_TIMESTAMPSQL_TYPE_TIME
#10.6.错误码
要获取错误码 可以使用SQLGetDiagRec()函数它会返回一个ANSI SQL标准定义的错误码字符串比如
SQLHENV env;
SQLAllocHandle(SQL_HANDLE_ENV, SQL_NULL_HANDLE, env);SQLSetEnvAttr(env, SQL_ATTR_ODBC_VERSION, reinterpret_castvoid*(SQL_OV_ODBC3), 0);SQLHDBC dbc;
SQLAllocHandle(SQL_HANDLE_DBC, env, dbc);SQLCHAR connectStr[] DRIVER{Apache Ignite};SERVERlocalhost;PORT10800;SCHEMAPerson;;
SQLDriverConnect(dbc, NULL, connectStr, SQL_NTS, 0, 0, 0, SQL_DRIVER_COMPLETE);SQLAllocHandle(SQL_HANDLE_STMT, dbc, stmt);SQLCHAR query[] SELECT firstName, lastName, resume, salary FROM Person;
SQLRETURN ret SQLExecDirect(stmt, query, SQL_NTS);if (ret ! SQL_SUCCESS)
{SQLCHAR sqlstate[7] ;SQLINTEGER nativeCode;SQLCHAR message[1024];SQLSMALLINT reallen 0;int i 1;ret SQLGetDiagRec(SQL_HANDLE_STMT, stmt, i, sqlstate,nativeCode, message, sizeof(message), reallen);while (ret ! SQL_NO_DATA){std::cout sqlstate : message;i;ret SQLGetDiagRec(SQL_HANDLE_STMT, stmt, i, sqlstate,nativeCode, message, sizeof(message), reallen);}
}下表中列出了所有Ignite目前支持的错误码该列表未来可能会扩展
错误码描述01S00无效连接串属性01S02驱动程序不支持指定的值并替换了一个类似的值08001驱动接入集群失败08002连接已经建立08003未知原因导致的连接处于关闭状态08004连接被集群踢出08S01连接失败22026字符串长度与数据执行对话框不匹配23000违反完整性约束比如主键重复、主键为空等等24000无效的游标状态42000请求的语法错误42S01表已经存在42S02表不存在42S11索引已经存在42S12索引不存在42S21列已经存在42S22列不存在HY000一般性错误具体看错误消息HY001内存分配错误HY003无效的应用缓冲区类型HY004无效的SQL数据类型HY009无效的空指针使用HY010函数调用顺序错误HY090无效的字符串和缓冲区长度比如长度为负或者为0HY092可选类型超范围HY097列类型超范围HY105无效的参数类型HY106获取类型超范围HYC00特性未实现IM001函数不支持
#11.多版本并发控制
警告
MVCC当前处于测试阶段。
#11.1.概述
配置为TRANSACTIONAL_SNAPSHOT原子化模式的缓存支持SQL事务以及键-值事务并且为两种类型的事务开启了多版本并发控制MVCC。
#11.2.多版本并发控制
多版本并发控制是一种在多用户并发访问时控制数据一致性的方法MVCC实现了快照隔离保证确保每个事务总是看到一致的数据快照。
每个事务在开始时会获得一个数据的一致性快照并且只能查看和修改该快照中的数据。当事务更新一个条目时Ignite验证其它事务没有更新该条目并创建该条目的一个新的版本新版本只有在事务成功提交时才对其它事务可见。如果该条目已被更新当前事务会失败并抛出异常关于如何处理更新冲突请参见下面的并发更新章节的介绍。
快照不是物理快照而是由MVCC协调器生成的逻辑快照MVCC协调器是协调集群中的事务活动的集群节点。协调器跟踪所有活动的事务并在每个事务完成时得到通知。启用MVCC的缓存的所有操作都需要从协调器请求一个数据的快照。
#11.3.开启MVCC
要为缓存开启MVCC需要在缓存配置中使用TRANSACTIONAL_SNAPSHOT原子化模式如果使用CREATE TABLE命令创建表则需要在该命令的WITH子句中将原子化模式作为参数传进去。
XMLSQL
CREATE TABLE Person WITH ATOMICITYTRANSACTIONAL_SNAPSHOT提示
TRANSACTIONAL_SNAPSHOT模式只支持默认的并发模型PESSIMISTIC和默认的隔离级别REPEATABLE_READ具体可以看上面的并发模型和隔离级别章节。
#11.4.并发更新
在一个事务中如果一个条目先被读取然后被更新那么就有一种可能性即另一个事务可能在两个操作之间切入然后首先更新该条目这时当第一个事务试图更新该条目时就会抛出异常然后该事务会被标记为只能回滚这时开发者就需要进行事务重试。
那么怎么知道发生了冲突呢
如果使用了Java的事务API会抛出CacheException异常异常信息为Cannot serialize transaction due to write conflict (transaction is marked for rollback)并且Transaction.rollbackOnly标志为true如果通过JDBC/ODBC驱动执行了SQL事务那么会得到SQLSTATE:40001错误代码。
JavaJDBCC#/.NETC
for(int i 1; i 5 ; i) {try (Transaction tx Ignition.ignite().transactions().txStart()) {System.out.println(attempt # i , value: cache.get(1));try {cache.put(1, new value);tx.commit();System.out.println(attempt # i succeeded);break;} catch (CacheException e) {if (!tx.isRollbackOnly()) {// Transaction was not marked as rollback only,// so its not a concurrent update issue.// Process the exception here.break;}}}
}#11.5.限制
#11.5.1.跨缓存事务
TRANSACTIONAL_SNAPSHOT模式是缓存级的并且事务中涉及的缓存不允许有不同的原子化模式因此如果希望在一个SQL事务中覆盖多个表则必须使用TRANSACTIONAL_SNAPSHOT模式创建所有表。
#11.5.2.嵌套事务
通过JDBC/ODBC连接参数Ignite支持三种模式来处理SQL的嵌套事务
JDBC连接串
jdbc:ignite:thin://127.0.0.1/?nestedTransactionsModeCOMMIT当在一个事务中出现了嵌套事务系统的行为依赖于nestedTransactionsMode参数
ERROR如果发生了嵌套事务会抛出错误包含事务会回滚这是默认的行为COMMIT包含事务会被提交嵌套事务开始并且在遇到COMMIT语句时会提交包含事务中的其余部分会作为隐式事务执行IGNORE不要使用这个模式嵌套事务的开始会被忽略嵌套事务中的语句会作为包含事务的一部分来执行然后所有的变更会随着嵌套事务的提交而提交包含事务中的其余部分会作为隐式事务执行。
#11.5.3.持续查询
如果在开启了MVCC的缓存上使用持续查询那么有些限制需要知道
当接收到更新事件时在MVCC协调器得知更新之前后续读取已更新键的操作可能会在一段时间内返回旧值。这是因为更新事件是从更新键的节点发送的并且是在键更新后立即发送的。这时MVCC协调器可能不会立即感知该更新因此随后的读取可能会在这段时间内返回过时的信息。当使用持续查询时每个节点单个事务可以更新的键数量是有限制的。更新后的值保存在内存中如果更新太多节点可能没有足够的内存来保存所有对象。为了避免内存溢出错误每个事务在一个节点上只能最多更新20000个键默认值。如果超过此值事务将抛出异常并回滚。可以通过指定IGNITE_MVCC_TX_SIZE_CACHING_THRESHOLD系统属性来修改该值。
#11.5.4.其它的限制
开启MVCC的缓存下面的特性是不支持的这些限制后续的版本可能会解决
近缓存过期策略事件缓存拦截器外部存储堆内缓存显式锁localEvict()和localPeek()方法。
