赞
踩
最近有一个需求,是将一些信息发送给指定的kafka(也就是咱们作为生产者)。在发送的时候,需要提供用户名和密码接收方才能成功接收。如果密码是明文,我们通常设置配置如下:
- properties.put("security.protocol", "PLAINTEXT");
- properties.put("sasl.mechanism", "PLAIN");
- properties.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"username\" password=\"123456789\";");
但是这次我们使用的密码是为MD5加密的,所以配置改为:
- properties.put("security.protocol", "SASL_PLAINTEXT");
- properties.put("sasl.mechanism", "SCRAM-SHA-512");
- properties.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"username\" password=\"7ed7845e45765fb75b045273e044cd21\";");
这种配置方法亲测有效,完整的kafka生产者代码如下:
- public void sendKafka(String str) {
- String BROKERS = "134.108.84.210:8423,134.108.84.94:8423,134.108.85.65:8423";//公共接入点地址
- Properties properties = new Properties();
- properties.put("bootstrap.servers", BROKERS);
- properties.put("acks", "1");// -1 all ack;1 leader ack;0 none ack
- properties.put("retries", 3);
- properties.put("batch.size", 262144);
- properties.put("linger.ms", 10);
- properties.put("buffer.memory", 67108864);
- properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- properties.put("client.id", "producer-olt_" + System.currentTimeMillis());
- properties.put("group.id","zj_olt_alarm");
- properties.put("security.protocol", "SASL_PLAINTEXT");
- properties.put("sasl.mechanism", "SCRAM-SHA-512");
- properties.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"username\" password=\"7ed7845e45765fb75b045273e044cd21\";");
- //构造Producer对象,注意,该对象是线程安全的
- //一般来说,一个进程内一个Producer对象即可,如果想提高性能,可构造多个对象,但最好不要超过5个
- KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
- try {
- ProducerRecord<String, String> record = new ProducerRecord<String, String>("TOPIC",str);
- producer.send(record, new Callback() {
- @Override
- public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
- if (exception != null) {
- //todo log and retry
- System.out.println("================================================");
- System.out.printf("send exception. %s%n", exception.getMessage());
- return;
- }
- //todo business process...
- System.out.println("================================================");
- System.out.printf("send message success. topic=%s,offset=%s,partition=%d %n",
- recordMetadata.topic(), recordMetadata.offset(), recordMetadata.partition());
- }
- });
- // System.out.println("推送成功");
- } catch (Exception e) {
- e.printStackTrace();
- logger.error("{kafka发送失败}",e);
- }
- }
以上方法提供参考,如有错误还请指出,共同进步
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。