應用程序能夠通過 Producer API 發送數據流至 Event Store 集群的 Topics。

KafkaPropertyUtil.java

public class KafkaPropertyUtil \{

    /**

    * 獲取配置文件中的參數

    * @param configPath

     * @return

    */

 public static Properties loadProperties(String configPath) \{

        InputStream in = null;

        Properties properties = null;

        try \{

            in = new BufferedInputStream(new
FileInputStream(configPath));

            properties = new Properties();

            properties.load(in);

        } catch (FileNotFoundException e) \{

            e.printStackTrace();

        } catch (IOException e) \{

            e.printStackTrace();

        } finally \{

            try \{

                in.close();

            } catch (IOException e) \{

                e.printStackTrace();

            }

        }

        return properties;

    }

}

SafeProducer.java

public class SafeProducer \{

    String topic;

    String propertyFilePath;

    String safeConfigFilePath;



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

        this.topic = topic;

        this.propertyFilePath = propertyFilePath;

        this.safeConfigFilePath = safeConfigFilePath;

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

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

    }



    /**

 * 發送數據

 */

 public void sendMessage() \{

        //獲取生產者參數

 Properties producerProperties =
KafkaPropertyUtil.loadProperties(propertyFilePath);

        KafkaProducer producer = NEW KafkaProducer(producerProperties);

        INT messageNo = 1;

        final INT COUNT = 5;



        while(messageNo < COUNT) \{

            String KEY = String.valueOf(messageNo);

            String DATA = String.format("hello KafkaProducer message
%s", KEY);

            try \{

                producer.send(NEW ProducerRecord<String, String>(topic,
DATA));

                System.out.println(DATA + " 發送成功");

            } catch (Exception e) \{

                e.printStackTrace();

            }

            messageNo++;

        }

        producer.close();

    }

}

測試文件:ProducerTest.java

 public class ProducerTest \{

 public void producerTest() \{

      String topic = "test_security";

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

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

      SafeProducer safeProducer = NEW SafeProducer(topic,
propertyFilePath, safeConfigFilePath);

      safeProducer.sendMessage();

   }

}

杠杆炒股的app合法吗-财经网最新消息股市-【东方资本】,做中线如何选股,炒股配资网站约选配资,业绩好的电力股