应用程序能够通过Consumer API从Event Store集群的Topics读取数据流。

SafeConsumer.java

public class SafeConsumer \{

    String topic;

    String propertyFilePath;

    String safeConfigFilePath;



    public SafeConsumer(String topic, String propertyFilePath, String
safeConfigFilePath) \{

        this.propertyFilePath = propertyFilePath;

        this.safeConfigFilePath = safeConfigFilePath;

        this.topic = topic;

        System.setProperty("java.security.auth.login.config",
safeConfigFilePath + "/jaas.conf");

        System.setProperty("java.security.krb5.conf", safeConfigFilePath
+ "/krb5.conf");

    }



    public void consumeMsg() \{

        Properties properties =
KafkaPropertyUtil.loadProperties(propertyFilePath);//KafkaPropertyUtil.loadProperties()见3.1.4
KafkaPropertyUtil.java

        KafkaConsumer<String, String> kafkaConsumer = NEW
KafkaConsumer<String, String>(properties);

        kafkaConsumer.subscribe(Arrays.asList(topic));



        while (TRUE) \{

            ConsumerRecords<String, String> records =
kafkaConsumer.poll(Duration.ofMillis(100));

            FOR (ConsumerRecord<String, String> record : records) \{

                System.out.println("--------------------------------");

                System.out.printf("offset=%d, value=%s",
record.offset(), record.value());

                System.out.println();

            }

        }

    }

}

测试文件:ConsumerTest.java

public class ConsumerTest \{

 public void consumerTest() \{

 String topic = "test_security";

 String propertyFilePath = "xxx/xxx/consumer.properties";

 String safeConfigFilePath = "/etc/eventstore1/conf";

 SafeConsumer safeConsumer = NEW SafeConsumer(topic, propertyFilePath,
safeConfigFilePath);

 safeConsumer.consumeMsg();

 }

}