java - Hazelcast EntryProcessor ClassNotFoundException

Question

我正在使用 hazelcast 缓存版本 3.11.2 来存储和操作 IMap 中的值。

对于 IMap 操作，我使用从 AbstractEntryProcessor 扩展的 EntryProcessor 类。

我观察到，在 IMap 代码上调用 put 或 putIfAbsent 方法时工作正常。

但是，在使用 EntryProcessor 操作 IMap 条目时，我收到 com.hazelcast.nio.serialization.HazelcastSerializationException: java.lang.ClassNotFoundException。

EntryProcessor 类

public class CountsEntryProcessor extends AbstractEntryProcessor<String, LocalAndRemoteCount> {

private static final long serialVersionUID = 7155522117375911439L;
long localCount;

public CountsEntryProcessor() {
}

public CountsEntryProcessor(long localCount) {
    this.localCount = localCount;
}

@Override
public Object process(Map.Entry<String, LocalAndRemoteCount> entry) {
    System.out.println("in entry processor");
    if(entry.getValue() != null){
        LocalAndRemoteCount localAndRemoteCount = entry.getValue();
        entry.setValue(new LocalAndRemoteCount(localAndRemoteCount.getLocalCount() + localCount, localAndRemoteCount.getRemoteCount()));
    }
    return null;
}
}

LocalAndRemoteCount 类

public class LocalAndRemoteCount implements Serializable{
private static final long serialVersionUID = 1085817105782061822L;
private long localCount;
private long remoteCount;

public LocalAndRemoteCount(long localCount, long remoteCount) {
    this.localCount = localCount;
    this.remoteCount = remoteCount;
}

public long incrementAndGetLocalCount() {
    return ++localCount;
}

public long addAndGetCount(long delta) {
    return localCount=localCount+delta;
}
public long getLocalCount() {
    return localCount;
}

public long getRemoteCount() {
    return remoteCount;
}

public long getTotalCount() {
    return getLocalCount() + getRemoteCount();
}

public void addRemoteLocalCount(long remoteMapLocalCount){
    remoteCount = remoteCount + remoteMapLocalCount;
}

@Override
public String toString(){
    return JsonObjectParser.getJson(LocalAndRemoteCount.class, this);
}

}

调用入口处理器

countsMap 是我的 Java 代码中的本地 Hashmap，它存储很少的值。

for (Map.Entry<String, Long> entry : countsMap.entrySet()) {
   publisher.getHazelcast().getCacheClient().getMap("ratelimiter.policies.ingest.map").putIfAbsent(entry.getKey(), new LocalAndRemoteCount(entry.getValue(), 0)); //Works fine
publisher.getHazelcast().getCacheClient().getMap("ratelimiter.policies.ingest.map").executeOnKey(entry.getKey(), new CountsEntryProcessor(entry.getValue())); //throws exception
}

异常堆栈跟踪

com.hazelcast.nio.serialization.HazelcastSerializationException: java.lang.ClassNotFoundException: commons.producer.jobs.CountsEntryProcessor
at com.hazelcast.internal.serialization.impl.JavaDefaultSerializers$JavaSerializer.read(JavaDefaultSerializers.java:86)
at com.hazelcast.internal.serialization.impl.JavaDefaultSerializers$JavaSerializer.read(JavaDefaultSerializers.java:75)
at com.hazelcast.internal.serialization.impl.StreamSerializerAdapter.read(StreamSerializerAdapter.java:48)
at com.hazelcast.internal.serialization.impl.AbstractSerializationService.toObject(AbstractSerializationService.java:187)
at com.hazelcast.client.impl.protocol.task.map.MapExecuteOnKeyMessageTask.prepareOperation(MapExecuteOnKeyMessageTask.java:42)
at com.hazelcast.client.impl.protocol.task.AbstractPartitionMessageTask.processMessage(AbstractPartitionMessageTask.java:62)
at com.hazelcast.client.impl.protocol.task.AbstractMessageTask.initializeAndProcessMessage(AbstractMessageTask.java:123)
at com.hazelcast.client.impl.protocol.task.AbstractMessageTask.doRun(AbstractMessageTask.java:111)
at com.hazelcast.client.impl.protocol.task.AbstractMessageTask.run(AbstractMessageTask.java:101)
at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:161)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:159)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:127)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.run(OperationThread.java:110)
at ------ submitted from ------.(Unknown Source)
at com.hazelcast.client.spi.impl.ClientInvocationFuture.resolveAndThrowIfException(ClientInvocationFuture.java:96)
at com.hazelcast.client.spi.impl.ClientInvocationFuture.resolveAndThrowIfException(ClientInvocationFuture.java:33)
at com.hazelcast.spi.impl.AbstractInvocationFuture.get(AbstractInvocationFuture.java:162)
at com.hazelcast.client.spi.ClientProxy.invokeOnPartition(ClientProxy.java:225)
at com.hazelcast.client.spi.ClientProxy.invoke(ClientProxy.java:219)
at com.hazelcast.client.proxy.ClientMapProxy.executeOnKeyInternal(ClientMapProxy.java:1447)
at com.hazelcast.client.proxy.ClientMapProxy.executeOnKey(ClientMapProxy.java:1440)
at commons.producer.jobs.MicroBatchJob.execute(MicroBatchJob.java:61)
at org.quartz.core.JobRunShell.run(JobRunShell.java:202)
at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:573)
Caused by: java.lang.ClassNotFoundException: commons.producer.jobs.CountsEntryProcessor
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at com.hazelcast.nio.ClassLoaderUtil.tryLoadClass(ClassLoaderUtil.java:288)
at com.hazelcast.nio.ClassLoaderUtil.loadClass(ClassLoaderUtil.java:252)
at com.hazelcast.nio.IOUtil$ClassLoaderAwareObjectInputStream.resolveClass(IOUtil.java:646)
at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1868)
at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1751)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2042)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431)
at com.hazelcast.internal.serialization.impl.JavaDefaultSerializers$JavaSerializer.read(JavaDefaultSerializers.java:82)
at com.hazelcast.internal.serialization.impl.JavaDefaultSerializers$JavaSerializer.read(JavaDefaultSerializers.java:75)
at com.hazelcast.internal.serialization.impl.StreamSerializerAdapter.read(StreamSerializerAdapter.java:48)
at com.hazelcast.internal.serialization.impl.AbstractSerializationService.toObject(AbstractSerializationService.java:187)
at com.hazelcast.client.impl.protocol.task.map.MapExecuteOnKeyMessageTask.prepareOperation(MapExecuteOnKeyMessageTask.java:42)
at com.hazelcast.client.impl.protocol.task.AbstractPartitionMessageTask.processMessage(AbstractPartitionMessageTask.java:62)
at com.hazelcast.client.impl.protocol.task.AbstractMessageTask.initializeAndProcessMessage(AbstractMessageTask.java:123)
at com.hazelcast.client.impl.protocol.task.AbstractMessageTask.doRun(AbstractMessageTask.java:111)
at com.hazelcast.client.impl.protocol.task.AbstractMessageTask.run(AbstractMessageTask.java:101)
at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:161)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:159)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:127)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.run(OperationThread.java:110)

Hazelcast.xml 文件中是否需要更改任何配置才能使用 EntryProcessor？我在这里错过了什么吗？

Answer 1

我唯一缺少的是 - 必须将包含入口处理器类的 jar 复制到 user-lib 目录中。