我有一个服务器节点和2个客户端节点。第一个客户端将条目放入缓存（我的缓存），然后
另一种使用连续查询来执行某些操作。以下是连续查询的代码段：
IgniteCache<Integer, MyDomain> cache = ignite.cache("my-cache");
ContinuousQuery<Integer, MyDomain> qry = new ContinuousQuery<>();
qry.setLocalListener(evts -> {
for (CacheEntryEvent<? extends Integer, ? extends MyDomain> e : evts) {
doSomething();
}
});
qry.setRemoteFilter(evt -> doSomething(evt));
cache.query(qry);
当我尝试启动群集时，在第二个客户端节点上出现以下异常：
[19:03:36] (err) Failed to notify listener: o.a.i.i.util.future.GridFutureChainListener@2e28f5a0class org.apache.ignite.IgniteException: Failed to initialize a remote filter.
at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.lambda$register$2bf956f5$1(CacheContinuousQueryHandler.java:337)
at org.apache.ignite.internal.util.future.GridFutureChainListener.applyCallback(GridFutureChainListener.java:78)
at org.apache.ignite.internal.util.future.GridFutureChainListener.apply(GridFutureChainListener.java:70)
at org.apache.ignite.internal.util.future.GridFutureChainListener.apply(GridFutureChainListener.java:30)
at org.apache.ignite.internal.util.future.GridFutureAdapter.notifyListener(GridFutureAdapter.java:385)
at org.apache.ignite.internal.util.future.GridFutureAdapter.unblock(GridFutureAdapter.java:349)
at org.apache.ignite.internal.util.future.GridFutureAdapter.unblockAll(GridFutureAdapter.java:337)
at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:497)
at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:476)
at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:464)
at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.p2pUnmarshal(CacheContinuousQueryHandler.java:1231)
at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.p2pUnmarshal(CacheContinuousQueryHandler.java:1203)
at org.apache.ignite.internal.processors.continuous.GridContinuousProcessor$8.run(GridContinuousProcessor.java:689)
at org.apache.ignite.internal.util.IgniteUtils.wrapThreadLoader(IgniteUtils.java:6864)
at org.apache.ignite.internal.processors.closure.GridClosureProcessor$1.body(GridClosureProcessor.java:827)
at org.apache.ignite.internal.util.worker.GridWorker.run(GridWorker.java:120)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: class org.apache.ignite.internal.IgniteDeploymentCheckedException: Failed to obtain deployment for class: com.abc.service.MyService$$Lambda$286/1896828359
at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryDeployableObject.unmarshal(CacheContinuousQueryDeployableObject.java:95)
at org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.p2pUnmarshal(CacheContinuousQueryHandler.java:1228)
... 8 more
最佳答案
解决问题的方法如下
创建了两个类：
public class CacheEntryEventFilterImpl implements CacheEntryEventFilter<Integer, MyDomain> {
@Override
public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends MyDomain> event)
throws CacheEntryListenerException {
return doSomething();
}
}
public class FactoryImpl implements Factory<CacheEntryEventFilter<Integer, MyDomain>> {
private static final long serialVersionUID = 881669537085864136L;
@Override
public CacheEntryEventFilter<Integer, MyDomain> create() {
return new CacheEntryEventFilterImpl();
}
}
在每个节点上都有上述类。
最后用
qry.setRemoteFilter(evt -> doSomething(evt));替换qry.setRemoteFilterFactory(new FactoryImpl());关于java - 点燃连续查询:无法初始化远程过滤器，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/59872877/