在不启用p2p类加载的情况下使apache ignite连续查询工作

Question

我一直试图让我的ignite连续查询代码在不将对等类加载设置为Enabled的情况下工作。但是，我发现代码不起作用。我试着调试，发现对cache.query（qry）的调用出错，并出现消息“failed to marshal custom event”错误。当我启用对等类加载时，代码按预期工作。有没有人可以提供指导，说明我如何在不加载对等类的情况下工作？下面是调用连续查询的代码段。

   public  void subscribeEvent(IgniteCache<String,String> cache,String inKeyStr,ServerWebSocket websocket ){
    System.out.println("in thread "+Thread.currentThread().getId()+"-->"+"subscribe event");
     //ArrayList<String> inKeys = new ArrayList<String>(Arrays.asList(inKeyStr.split(",")));

     ContinuousQuery<String, String> qry = new ContinuousQuery<>();

    /****
     * Continuous Query Impl
     */

    inKeys = ","+inKeyStr+",";
    qry.setInitialQuery(new ScanQuery<String, String>((k, v) -> inKeys.contains(","+k+",")));
    qry.setTimeInterval(1000);
    qry.setPageSize(1);

    // Callback that is called locally when update notifications are received.
  //  Factory<CacheEntryEventFilter<String, String>> rmtFilterFactory = new com.ccx.ignite.cqfilter.FilterFactory().init(inKeyStr);      

    qry.setLocalListener(new CacheEntryUpdatedListener<String, String>() {

          @Override public void onUpdated(Iterable<CacheEntryEvent<? extends String, ? extends String>> evts) {
            for (CacheEntryEvent<? extends String, ? extends String> e : evts)
            {

                          System.out.println("websocket locallsnr data in thread "+Thread.currentThread().getId()+"-->"+"key=" + e.getKey() + ", val=" + e.getValue());
                          try{
                          websocket.writeTextMessage("key=" + e.getKey() + ", val=" + e.getValue());
                          }
                          catch (Exception e1){
                              System.out.println("exception local listener "+e1.getMessage());
                              qry.setLocalListener(null) ;                }

            }

          }
        } );

   qry.setRemoteFilterFactory( new com.ccx.ignite.cqfilter.FilterFactory().init(inKeys));

    try{        

          cur = cache.query(qry);
          for (Cache.Entry<String, String> e : cur)
         { 
           System.out.println("websocket initialqry data in thread "+Thread.currentThread().getId()+"-->"+"key=" + e.getKey() + ", val=" + e.getValue());
           websocket.writeTextMessage("key=" + e.getKey() + ", val=" + e.getValue());
         }
    }
    catch (Exception e){
          System.out.println("exception cache.query "+e.getMessage());

          }        
    }

下面是远程筛选器类，我已经将它制作成一个自包含的jar并推送到ignite的libs文件夹中，这样服务器节点就可以获取它

public class FilterFactory
{

public Factory<CacheEntryEventFilter<String, String>> init(String inKeyStr ){
System.out.println("factory init called jun22 ");

return new Factory <CacheEntryEventFilter<String, String>>() {
    private static final long serialVersionUID = 5906783589263492617L;

        @Override public CacheEntryEventFilter<String, String> create() {
            return new CacheEntryEventFilter<String, String>() {
                @Override  public boolean evaluate(CacheEntryEvent<? extends String, ? extends String> e) {
                    //List inKeys = new ArrayList<String>(Arrays.asList(inKeyStr.split(",")));
                    System.out.println("inside remote filter factory ");
                    String inKeys = ","+inKeyStr+",";

                    return inKeys.contains(","+e.getKey()+",");
                }
            };
        }
    };
}
}

Answer 1

您需要在拓扑中的所有节点上显式部署CQ类（特别是远程筛选器）。只需使用它们创建一个JAR文件，并在启动节点之前将其放入libs文件夹。