使用 Infinispan Cache 和 computeIfAbsent 的无效 lambda 反序列化

Question

我正在玩弄一个基本的 infinispan 集群，我遇到了一个令人费解的错误。

我基本上是在实现一个共享映射，只包含一个整数

这是我的服务代码

package sandbox.infinispan.test.service;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.inject.Named;

import org.infinispan.Cache;

@Named("useThisOne")
@ApplicationScoped
public class CounterService implements ICounterService {

    private static final String KEY = "key";

    @Inject
    private Cache<String, Integer> cache;

    @Override
    public void inc(final int amount) {
        this.cache.put(KEY, Integer.valueOf(this.get() + amount));
    }

    @Override
    public int get() {
        return this.cache.computeIfAbsent(KEY, k -> Integer.valueOf(0)).intValue();
    }
}

缓存由以下内容生成：

package sandbox.infinispan.test.config;

import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.context.Dependent;
import javax.enterprise.inject.Produces;

import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.Configuration;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfiguration;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;

@Dependent
class CacheProvider {

    @Produces
    @ApplicationScoped
    private EmbeddedCacheManager defaultClusteredCacheManager() {
        final GlobalConfiguration g = new GlobalConfigurationBuilder() //
                .clusteredDefault() //
                .transport() //
                .nodeName(this.getNodeName()) //
                .clusterName("infinispanTestCluster") //
                .build();
        final Configuration cfg = new ConfigurationBuilder() //
                .clustering() //
                .cacheMode(CacheMode.REPL_SYNC) ///
                .build();
        return new DefaultCacheManager(g, cfg);
    }
}

当集群中至少有两台服务器时，computeIfAbsent 失败

15:48:50,253 ERROR [org.infinispan.interceptors.impl.InvocationContextInterceptor] (jgroups-7,myhostname-14393) ISPN000136: Error executing command ComputeIfAbsentCommand, writing keys [key]: org.infinispan.remoting.RemoteException: ISPN000217: Received exception from otherhostname-44445, see cause for remote stack trace

深入到：

Caused by: java.lang.NoSuchMethodException: sandbox.infinispan.test.service.CounterService.$deserializeLambda$(java.lang.invoke.SerializedLambda)

最后到：

Caused by: java.lang.IllegalArgumentException: Invalid lambda deserialization
        at sandbox.infinispan.test.service.CounterService.$deserializeLambda$(CounterService.java:10)
            ... 68 more
Caused by: an exception which occurred:
        in object of type java.lang.invoke.SerializedLambda

如果我将我漂亮的时尚代码重写成丑陋的后续代码，它就可以工作。

@Override
public int get() {
    Integer value = this.cache.get(KEY);
    if (value == null) {
        value = Integer.valueOf(0);
        this.cache.put(KEY, value);
    }
    return value.intValue();
}

现在如何使用漂亮的 computeIfAbsent 做事方式？

Eclipse 2018-12，WildFly 14，java 10 on of the dev member of the cluster, CentOs 7, OpenJdk 10, WildFly 14 on the remote cluster member.

感谢您的帮助

已解决（有点）

感谢我在这里得到的帮助，我将 lambda 转换为内部 class :

static class OhWell implements Serializable {

    static Integer zero(final String t) {
        return Integer.valueOf(0);
    }
}

@Override
public int get() {
    return this.cache.computeIfAbsent(KEY, OhWell::zero).intValue();
}

它现在可以工作了，但它不如整洁的 lambda 好得多。所以我会坚持老式的方式——除非有人能想到更好的方法。

进一步结果：

以下 static 内部 class 和 static 方法有效

static class StaticOhWell implements Serializable {

    static Integer apply(final String t) {
        return Integer.valueOf(0);
    }
}

@Override
public int get() {
    return this.cache.computeIfAbsent(KEY, StaticOhWell::apply).intValue();
}

以下 非静态 内部 class 与 非静态 方法失败：

class NotStaticOhWell implements SerializableFunction<String, Integer> {

    @Override
    public Integer apply(final String t) {
            return Integer.valueOf(0);
    }
}

@Override
public int get() {
    return this.cache.computeIfAbsent(KEY, new NotStaticOhWell()::apply).intValue();
}

失败并显示此错误消息 NotSerializableException: org.infinispan.cache.impl.EncoderCache:

13:41:29,221 ERROR [org.infinispan.interceptors.impl.InvocationContextInterceptor] (default task-1) ISPN000136: Error executing command ComputeIfAbsentCommand, writing keys [value]: org.infinispan.commons.marshall.NotSerializableException: org.infinispan.cache.impl.EncoderCache
    Caused by: an exception which occurred:
    in field sandbox.infinispan.test.service.CounterService.cache
    in object sandbox.infinispan.test.service.CounterService@4612a6c3
    in field sandbox.infinispan.test.service.CounterService$NotStaticOhWell.this[=19=]
    in object sandbox.infinispan.test.service.CounterService$NotStaticOhWell@4effd362
    in field java.lang.invoke.SerializedLambda.capturedArgs
    in object java.lang.invoke.SerializedLambda@e62f08a
    in object sandbox.infinispan.test.service.CounterService$$Lambda95/1060417313@174a143b

最后的话(?)

使用“静态 lambda”（static 内部 class 实现 SerializableFunction 接口）也有效

static class StaticSerializableFunction implements SerializableFunction<String, Integer> {

    @Override
    public Integer apply(final String t) {
        return Integer.valueOf(0);
    }
}

@Override
public int get() {
    return this.cache.computeIfAbsent(KEY, new StaticSerializableFunction()::apply).intValue();
}

获胜者是……

通过“transienting”使 class 实际上可序列化 Cache 允许简单地使用此 class。无需创建内部 class!

package sandbox.infinispan.test.service;

import java.io.Serializable;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.inject.Named;

import org.infinispan.Cache;

@Named("useThisOne")
@ApplicationScoped
public class CounterService implements ICounterService, Serializable {

    private static final String KEY = "value";

    @SuppressWarnings("cdi-ambiguous-dependency")
    @Inject
    private transient Cache<String, Integer> cache;

    @Override
    public void inc(final int amount) {
        this.cache.put(KEY, Integer.valueOf(this.get() + amount));
    }

    @Override
    public int get() {
        return this.cache.computeIfAbsent(KEY, this::zero).intValue();
    }

    private Integer zero(@SuppressWarnings("unused") final String unused) {
        return Integer.valueOf(0);
    }

    @Override
    public void reset() {
        this.cache.clear();
    }
}

谢谢大家！

Answer 1

根据，解串器需要实际代码可用。您确定您的应用程序已经在集群中的所有其他节点上启动（完全相同的版本，包括您的 lambda）吗？

computeIfAbsent() 将 lambda 直接发送到数据（因此使用一个 RPC 处理操作，而不是先获取值然后将其写入 'ugly code'）。在 WF 中，您的应用程序位于与 Infinispan 不同的 classloader（模块）中，这可能会导致问题。

您能否尝试将您的 lambda 重构为 class 并查看您是否遇到类似问题？我对 WF 不是很了解，因此对于 lambda 缺少的常规 classes 可能有缓解措施。

使用 Infinispan Cache 和 computeIfAbsent 的无效 lambda 反序列化

Invalid lambda deserialization with Infinispan Cache and computeIfAbsent

jgroups

infinispan

wildfly