没有提交 Ignite CacheStoreAdapter

Question

CacheStoreAdapter 出错。当它的方法 sessionEnd 被调用时，CacheStoreSessionResource 的连接总是关闭的。没有任何例外。但实际上事务没有提交并且数据库中没有任何更改。我的代码非常简单。一切都是按照原来的Ignite例子来做的。

客户端

IgniteCache<String, String> typeCache = ignite.getOrCreateCache("typeCache");
try (Transaction tx = ignite.transactions().txStart(TransactionConcurrency.PESSIMISTIC,
        TransactionIsolation.REPEATABLE_READ)) {
    typeCache.put("code", "name");
    tx.commit();
}
catch (Exception e) {
    log.error("ERROR. Put Type: " + type, e);
}

服务器：

public class CacheJdbcTypeStore extends CacheStoreAdapter<String, String> {

    @CacheStoreSessionResource
    private CacheStoreSession ses;

    @Override
    public void write(Cache.Entry<? extends String, ? extends String> entry) {
        String key = entry.getKey();
        String val = entry.getValue();

        try (Connection conn = connection(ses)) {
            try (PreparedStatement st = conn.prepareStatement(
                    "insert into t_type (code, name) values (?, ?)")) {
                st.setString(1, key);
                st.setString(2, val);
                st.executeUpdate();
            }
        }
        catch (Exception e) {
            throw new CacheWriterException("Failed to write Type " + val, e);
        }
    }

    @Override
    public void sessionEnd(boolean commit) {
        try (Connection conn = ses.attachment()) {
            if (conn != null && !conn.isClosed() && ses.isWithinTransaction()) {
                if (commit)
                    conn.commit();
                else
                    conn.rollback();
            }
        }
        catch (SQLException e) {
            throw new CacheWriterException("Failed to end store session of Type cache", e);
        }
    }

    private Connection connection(CacheStoreSession ses) throws Exception {
        if (ses.isWithinTransaction()) {
            Connection conn = ses.attachment();
    /************************************/
    /* Here conn always is closed. WHY???? */
    /* As result transaction is never commited !!!! */
    /************************************/
            if (conn == null || conn.isClosed()) {
                conn = openConnection(false);
                ses.attach(conn);
            }
            return conn;
        }
        else {
            return openConnection(true);
        }
    }

    // Opens JDBC connection.
    private Connection openConnection(boolean autocommit) throws Exception {
        Connection conn = DriverManager.getConnection(url);
        conn.setAutoCommit(autocommit);
        return conn;
    }

    @Override
    public String load(final String key) {
        return null;
    }

    @Override
    public void delete(Object key) {
    }

}

配置

<property name="cacheConfiguration">
    <list>
        <bean class="org.apache.ignite.configuration.CacheConfiguration">
            <property name="name" value="typeCache" />
            <property name="cacheMode" value="PARTITIONED" />
            <property name="atomicityMode" value="TRANSACTIONAL" />
            <property name="backups" value="1" />
            <property name="cacheStoreFactory">
                <bean class="javax.cache.configuration.FactoryBuilder"
                    factory-method="factoryOf">
                    <constructor-arg
                    value="ru.raiffeisen.cache.store.jdbc.CacheJdbcTypeStore" />
                </bean>
            </property>
            <property name="readThrough" value="true" />
            <property name="writeThrough" value="true" />
        </bean>
    </list>
</property>

Answer 1

您使用 try-with-resource 进行连接，因此每次离开此块时，连接都会关闭。

try (Connection conn = ses.attachment()) {}

我想你检查了这个实现：https://apacheignite.readme.io/docs/3rd-party-store#section-cachestore-example 但是，如您所见，它提到它不是事务性的。请检查 this 缓存存储实现作为事务缓存存储的示例

此外，要查看 conn 变量中的非空值，请尝试在事务中添加多个插入。

Answer 2

因此 Ignite 文档包含 jdbc-transactional CacheStoreAdapter 实现的错误示例。在上面的 'server' 代码中，我做了一些小改动以消除错误。

try {
    Connection conn = connection(ses);
        try {
            PreparedStatement st = conn.prepareStatement(
                "insert into t_type (code, name) values (?, ?)");
            st.setString(1, key);
            st.setString(2, val);
            st.executeUpdate();
        }
    }
    catch (Exception e) {
        throw new CacheWriterException("Failed to write Type " + val, e);
    }

没有提交 Ignite CacheStoreAdapter

No commit for Ignite CacheStoreAdapter

java

persistence

ignite