hazelcast-jet - Hazelcast Jet 0.5.1 服务器/客户端问题

Question

我正在尝试使用 Hazelcast Jet 0.5.1 运行服务器 - 客户端示例。

这个例子是从 hazelcast jet 下面的代码示例扩展而来的。

hazelcast-jet-0.5.1\code-samples\streaming\map-journal-source\src\main\java\RemoteMapJournalSource.java

这是一个场景

我有 1 个服务器程序和 2 个客户端程序。

服务器- 通常的 hazelcast 代码在 Hazelcast 映射中放置 100 个整数条目。它还为地图配置了事件日志。

客户端- 使用客户端配置连接到 hazelcast，从远程日志中读取条目并排放到 IList。

以下是观察结果

服务器将 100 个条目放入映射，但客户端 1 获得 200，客户端 2 获得 300 个条目（初始列表大小为 100）

我预计只有 100 个条目可用，并且条目将在这两个客户端之间分配。

示例代码如下。

服务器程序代码

public class RemoteMapJournalSourceSrv1 {

private static final String MAP_NAME = "map";
private static final String SINK_NAME = "list";

public static void main(String[] args) throws Exception {
    System.setProperty("remoteHz.logging.type", "log4j");

    Config hzConfig = getConfig();
    HazelcastInstance remoteHz = startRemoteHzCluster(hzConfig);

    try {

        IMap<Integer, Integer> map = remoteHz.getMap(MAP_NAME);
        System.out.println("*************** Initial Map  address " +  map.size() );
        
        for (int i = 0; i < 100; i++) {
            map.set(i, i);
        
        }
        System.out.println("***************map size "+map.size());
        
        TimeUnit.SECONDS.sleep(300);
        
        
    } finally {
        Hazelcast.shutdownAll();
    }

}

private static String getAddress(HazelcastInstance remoteHz) {
    Address address = remoteHz.getCluster().getLocalMember().getAddress();
    System.out.println("***************Remote address " + address.getHost() + ":" + address.getPort() );
    return address.getHost() + ":" + address.getPort();
}


private static HazelcastInstance startRemoteHzCluster(Config config) {
    HazelcastInstance remoteHz = Hazelcast.newHazelcastInstance(config);
    Hazelcast.newHazelcastInstance(config);
    return remoteHz;
}

private static Config getConfig() {
    Config config = new Config();
    // Add an event journal config for map which has custom capacity of 1000 (default 10_000)
    // and time to live seconds as 10 seconds (default 0 which means infinite)
    config.addEventJournalConfig(new EventJournalConfig().setEnabled(true)
                                                         .setMapName(MAP_NAME)
                                                         .setCapacity(10000)
                                                         .setTimeToLiveSeconds(100));
    return config;
}

客户 1 程序代码

 public class RemoteMapJournalSourceCL1 {

private static final String MAP_NAME = "map";
private static final String SINK_NAME = "list";

public static void main(String[] args) throws Exception {
    System.setProperty("remoteHz.logging.type", "log4j");

      JetInstance localJet = startLocalJetCluster();

    try {
        ClientConfig clientConfig = new ClientConfig();
        GroupConfig groupConfig = new GroupConfig();

        clientConfig.getNetworkConfig().addAddress("localhost:5701");
        clientConfig.setGroupConfig(groupConfig);
        
        IList list1 = localJet.getList(SINK_NAME);
        
        int size1 = list1.size();
        
        System.out.println("***************List Initial size "+size1);

        Pipeline p = Pipeline.create();
        p.drawFrom(Sources.<Integer, Integer, Integer>remoteMapJournal(MAP_NAME, clientConfig,
                e -> e.getType() == EntryEventType.ADDED, EventJournalMapEvent::getNewValue, false))
         .peek()
         .drainTo(Sinks.list(SINK_NAME));
        
        JobConfig jc= new JobConfig();
        jc.setProcessingGuarantee(ProcessingGuarantee.EXACTLY_ONCE);
        localJet.newJob(p,jc);

        TimeUnit.SECONDS.sleep(10);
        
        IList list = localJet.getList(SINK_NAME);
    
    
        int size = list.size();
        
            for(int j =0; j< size; j++ ){
            System.out.println("***************" + list.get(j));
        }
        
        System.out.println("***************Read " + size + " entries from remote map journal.");
    
        
    } finally {
        Hazelcast.shutdownAll();
        Jet.shutdownAll();
    }

}

private static String getAddress(HazelcastInstance remoteHz) {
    Address address = remoteHz.getCluster().getLocalMember().getAddress();
    System.out.println("***************Remote address " + address.getHost() + ":" + address.getPort() );
    return address.getHost() + ":" + address.getPort();
}

private static JetInstance startLocalJetCluster() {
      JetInstance localJet = Jet.newJetInstance();
    return localJet;
}

客户端 2 程序代码。

public class RemoteMapJournalSourceCL2 {

private static final String MAP_NAME = "map";
private static final String SINK_NAME = "list";

public static void main(String[] args) throws Exception {
    System.setProperty("remoteHz.logging.type", "log4j");

      JetInstance localJet = startLocalJetCluster();

    try {
        ClientConfig clientConfig = new ClientConfig();
        GroupConfig groupConfig = new GroupConfig();

        clientConfig.getNetworkConfig().addAddress("localhost:5701");
        clientConfig.setGroupConfig(groupConfig);
        
        IList list1 = localJet.getList(SINK_NAME);
        int size1 = list1.size();
        System.out.println("***************List Initial size "+size1);

        
        Pipeline p = Pipeline.create();
        p.drawFrom(Sources.<Integer, Integer, Integer>remoteMapJournal(MAP_NAME, clientConfig,
                e -> e.getType() == EntryEventType.ADDED, EventJournalMapEvent::getNewValue, false))
         .peek()
         .drainTo(Sinks.list(SINK_NAME));
        
        JobConfig jc= new JobConfig();
        jc.setProcessingGuarantee(ProcessingGuarantee.EXACTLY_ONCE);
        
        localJet.newJob(p,jc);

        TimeUnit.SECONDS.sleep(10);
        
        IList list = localJet.getList(SINK_NAME);
    
    
        int size = list.size();
        
            for(int j =0; j< size; j++ ){
            System.out.println("***************" + list.get(j));
        }
        
        System.out.println("***************Read " + size + " entries from remote map journal.");
    
        
    } finally {
        Hazelcast.shutdownAll();
        Jet.shutdownAll();
    }

}

private static String getAddress(HazelcastInstance remoteHz) {
    Address address = remoteHz.getCluster().getLocalMember().getAddress();
    System.out.println("***************Remote address " + address.getHost() + ":" + address.getPort() );
    return address.getHost() + ":" + address.getPort();
}

private static JetInstance startLocalJetCluster() {
      JetInstance localJet = Jet.newJetInstance();
    return localJet;
}

你能帮我解决这个问题吗？