這篇文章主要講解了“kafka javaAPI入庫程序的實(shí)現(xiàn)方法”,文中的講解內(nèi)容簡單清晰,易于學(xué)習(xí)與理解,下面請大家跟著小編的思路慢慢深入,一起來研究和學(xué)習(xí)“kafka javaAPI入庫程序的實(shí)現(xiàn)方法”吧!
十年的金堂縣網(wǎng)站建設(shè)經(jīng)驗,針對設(shè)計、前端、開發(fā)、售后、文案、推廣等六對一服務(wù),響應(yīng)快,48小時及時工作處理。網(wǎng)絡(luò)營銷推廣的優(yōu)勢是能夠根據(jù)用戶設(shè)備顯示端的尺寸不同,自動調(diào)整金堂縣建站的顯示方式,使網(wǎng)站能夠適用不同顯示終端,在瀏覽器中調(diào)整網(wǎng)站的寬度,無論在任何一種瀏覽器上瀏覽網(wǎng)站,都能展現(xiàn)優(yōu)雅布局與設(shè)計,從而大程度地提升瀏覽體驗。創(chuàng)新互聯(lián)公司從事“金堂縣網(wǎng)站設(shè)計”,“金堂縣網(wǎng)站推廣”以來,每個客戶項目都認(rèn)真落實(shí)執(zhí)行。
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.3.0</version> </dependency>
Properties props = new Properties(); props.put("acks", "all"); //保證所有副本接受到消息 props.put("bootstrap.servers", Config.ipList); //可設(shè)置多個 props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put("retries", "2"); KafkaProducer<byte[], byte[]> produce= new KafkaProducer<byte[], byte[]>(props);
kerberos是大數(shù)據(jù)平臺的安全認(rèn)證策略,可在項目啟動時先一步完成。這里介紹兩種實(shí)現(xiàn)方式。
指定認(rèn)證文件
//加載keberos配置文件 System.setProperty("java.security.krb5.conf", "/etc/krb5.conf"); //加載kerberos用戶文件 System.setProperty("java.security.auth.login.config", "/etc/kafka/conf/kafka_jaas.conf");
某些時候,考慮到用戶切換,不同機(jī)器,有不同的用戶信息,每個都要通過配置文件設(shè)置,比較麻煩,考慮使用java的啟動的臨時文件功能(主要是炫技——微笑)。
//加載keberos配置文件 System.setProperty("java.security.krb5.conf", "/etc/krb5.conf"); KafkaUtil.configureJAAS(Config.tabFile, Config.principal); //用戶和認(rèn)證文件 /** * 生成jaas.conf臨時文件 * @param keyTab tab認(rèn)證文件位置 * @param principal 認(rèn)證用戶 */ public static void configureJAAS(String keyTab, String principal) { String JAAS_TEMPLATE = "KafkaClient {\n" + "com.sun.security.auth.module.Krb5LoginModule required\n" + "useKeyTab=true\n" + "keyTab=\"%1$s\"\n" + "principal=\"%2$s\";\n" + "};"; String content = String.format(JAAS_TEMPLATE, keyTab, principal); File jaasConf = null; PrintWriter writer = null; try { jaasConf = File.createTempFile("jaas", ".conf"); writer = new PrintWriter(jaasConf); writer.println(content); } catch (IOException e) { e.printStackTrace(); } finally { if (writer != null) { writer.close(); } jaasConf.deleteOnExit(); } System.setProperty("java.security.auth.login.config", jaasConf.getAbsolutePath()); }
實(shí)際線上使用時,考慮到數(shù)據(jù)傳輸效率和穩(wěn)定性,要做以下優(yōu)化。
傳輸類為線程類,線程池管理,增加傳輸效率。
批量上傳數(shù)據(jù)。
添加Callback處理機(jī)制,避免數(shù)據(jù)丟失。
上傳線程類如下。
public class Performance extends Thread{ private final static Logger log = LoggerFactory.getLogger(Performance.class); private List<ProducerRecord<byte[], byte[]>> recordList; public Performance(List<ProducerRecord<byte[], byte[]>> recordList) { this.recordList=recordList; } /** *入庫測試方法 */ public static void test() { log.info("Kafka Tool Test"); try { /* parse args */ String topicName ="test40"; /*總發(fā)包數(shù)*/ long numRecords = 10000000000L; /*包大小*/ int recordSize = 1500; /*每次最多發(fā)送包數(shù)*/ int throughput = 10000000; Properties props = new Properties(); props.put("acks", "1"); props.put("bootstrap.servers","ip:6667,ip:6667"); props.put("sasl.kerberos.service.name", "kafka"); props.put("security.protocol", "SASL_PLAINTEXT"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); KafkaProducer<byte[], byte[]> producer = new KafkaProducer<byte[], byte[]>(props); /* 創(chuàng)建測試數(shù)據(jù) */ byte[] payload = new byte[recordSize]; Random random = new Random(0); for (int i = 0; i < payload.length; ++i) payload[i] = (byte) (random.nextInt(26) + 65); /*創(chuàng)建測試數(shù)據(jù)發(fā)送對象*/ ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[], byte[]>(topicName, payload); /*測試數(shù)據(jù)模型 包總數(shù)*/ Stats stats = new Stats(numRecords, 5000); /*啟動時間*/ long startMs = System.currentTimeMillis(); /*幫助生成者發(fā)送流量類 每次最多發(fā)送包數(shù) 時間*/ ThroughputThrottler throttler = new ThroughputThrottler(throughput, startMs); for (int i = 0; i < numRecords; i++) { long sendStartMs = System.currentTimeMillis(); Callback cb = stats.nextCompletion(sendStartMs, payload.length, stats,record.topic(),record.value()); producer.send(record, cb); if (throttler.shouldThrottle(i, sendStartMs)) { throttler.throttle(); } } /* 結(jié)束任務(wù) */ producer.close(); stats.printTotal(); } catch (Exception e) { log.info("Test Error:"+e); } } /** * 實(shí)際入庫方法 */ @Override public void run() { // log.info("Start To Send:"); super.run(); KafkaUtil kafkaUtil=new KafkaUtil(); KafkaProducer<byte[], byte[]> produce=kafkaUtil.create(); //總包數(shù) long size=recordList.size(); // size=10000000000L; /*每次最多發(fā)送包數(shù)*/ int throughput = 900000; // throughput = 10000000; /*測試數(shù)據(jù)模型 包總數(shù)*/ Stats stats = new Stats(size, 5000); /*啟動時間*/ long startMs = System.currentTimeMillis(); /*幫助生成者發(fā)送流量類 每次最多發(fā)送包數(shù) 時間*/ ThroughputThrottler throttler = new ThroughputThrottler(throughput, startMs); int i=0; for (ProducerRecord<byte[], byte[]> record:recordList) { long sendStartMs = System.currentTimeMillis(); //參數(shù)說明:發(fā)送數(shù)據(jù)時間 數(shù)據(jù)長度 數(shù)據(jù)模型類 Callback cb = stats.nextCompletion(sendStartMs, record.value().length, stats,record.topic(),record.value()); produce.send(record,cb); if (throttler.shouldThrottle(i, sendStartMs)) { throttler.throttle(); } i++; } produce.close(); // stats.printTotal(); // log.info("End to Send"); log.info("Finish Data To Send"); LogModel.sendNum++; } private static class Stats { private long start; private long windowStart; private int[] latencies; private int sampling; private int iteration; private int index; private long count; private long bytes; private int maxLatency; private long totalLatency; private long windowCount; private int windowMaxLatency; private long windowTotalLatency; private long windowBytes; private long reportingInterval; public Stats(long numRecords, int reportingInterval) { this.start = System.currentTimeMillis(); this.windowStart = System.currentTimeMillis(); this.index = 0; this.iteration = 0; this.sampling = (int) (numRecords / Math.min(numRecords, 500000)); this.latencies = new int[(int) (numRecords / this.sampling) + 1]; this.index = 0; this.maxLatency = 0; this.totalLatency = 0; this.windowCount = 0; this.windowMaxLatency = 0; this.windowTotalLatency = 0; this.windowBytes = 0; this.totalLatency = 0; this.reportingInterval = reportingInterval; } public void record(int iter, int latency, int bytes, long time) { this.count++; this.bytes += bytes; this.totalLatency += latency; this.maxLatency = Math.max(this.maxLatency, latency); this.windowCount++; this.windowBytes += bytes; this.windowTotalLatency += latency; this.windowMaxLatency = Math.max(windowMaxLatency, latency); if (iter % this.sampling == 0) { this.latencies[index] = latency; this.index++; } /* maybe report the recent perf */ if (time - windowStart >= reportingInterval) { printWindow(); newWindow(); } } public Callback nextCompletion(long start, int bytes, Stats stats,String topic,byte[] data) { Callback cb = new PerfCallback(this.iteration, start, bytes, stats,topic,data); this.iteration++; return cb; } /** * 傳輸效率反饋 */ public void printWindow() { long ellapsed = System.currentTimeMillis() - windowStart; double recsPerSec = 1000.0 * windowCount / (double) ellapsed; double mbPerSec = 1000.0 * this.windowBytes / (double) ellapsed / (1024.0 * 1024.0); System.out.printf("%d spend time,%d records sent, %.1f records/sec (%.2f MB/sec), %.1f ms avg latency, %.1f max latency.\n", ellapsed, windowCount, recsPerSec, mbPerSec, windowTotalLatency / (double) windowCount, (double) windowMaxLatency); } public void newWindow() { this.windowStart = System.currentTimeMillis(); this.windowCount = 0; this.windowMaxLatency = 0; this.windowTotalLatency = 0; this.windowBytes = 0; } /** * 傳輸效率 */ public void printTotal() { long elapsed = System.currentTimeMillis() - start; double recsPerSec = 1000.0 * count / (double) elapsed; double mbPerSec = 1000.0 * this.bytes / (double) elapsed / (1024.0 * 1024.0); int[] percs = percentiles(this.latencies, index, 0.5, 0.95, 0.99, 0.999); System.out.printf("%d spend time,%d records sent, %f records/sec (%.2f MB/sec), %.2f ms avg latency, %.2f ms max latency, %d ms 50th, %d ms 95th, %d ms 99th, %d ms 99.9th.\n", elapsed, count, recsPerSec, mbPerSec, totalLatency / (double) count, (double) maxLatency, percs[0], percs[1], percs[2], percs[3]); } private static int[] percentiles(int[] latencies, int count, double... percentiles) { int size = Math.min(count, latencies.length); Arrays.sort(latencies, 0, size); int[] values = new int[percentiles.length]; for (int i = 0; i < percentiles.length; i++) { int index = (int) (percentiles[i] * size); values[i] = latencies[index]; } return values; } } private static final class PerfCallback implements Callback { private final long start; private final int iteration; private final int bytes; private final Stats stats; private final String topic; private final byte[] data; public PerfCallback(int iter, long start, int bytes, Stats stats,String topic,byte[] data) { this.start = start; this.stats = stats; this.iteration = iter; this.bytes = bytes; this.topic=topic; this.data=data; } public void onCompletion(RecordMetadata metadata, Exception exception) { long now = System.currentTimeMillis(); int latency = (int) (now - start); this.stats.record(iteration, latency, bytes, now); if (exception != null){ ProducerRecord<byte[], byte[]> record=new ProducerRecord<byte[], byte[]>(topic,data); //將數(shù)據(jù)重新添加入數(shù)據(jù)隊列,二次上傳 ControlTask.recordList.add(record); log.error("Send Error And Second To Send",exception); } } } }
KafkaUtil.java
public class KafkaUtil { // private final static Logger log = LoggerFactory.getLogger(KafkaUtil.class); private KafkaProducer<byte[], byte[]> produce; /** * 創(chuàng)建連接 * @return */ public KafkaProducer<byte[], byte[]> create(){ Properties props = new Properties(); props.put("acks", "all"); props.put("bootstrap.servers", Config.ipList); props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); // props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 120000); //增加等待時間 props.put("retries", "2"); //kerbores安全認(rèn)證 if(Config.kerberos==0){ props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "GSSAPI"); props.put("sasl.kerberos.service.name", "kafka"); } produce = new KafkaProducer<byte[], byte[]>(props); return produce; } /** * 發(fā)送數(shù)據(jù) * @param record * @param cb */ public void send(ProducerRecord<byte[], byte[]> record,Callback cb){ produce.send(record,cb); } /** * 關(guān)閉連接 * @param produce */ public void close(){ produce.flush(); produce.close(); } /** * 生成jaas.conf臨時文件 * @param keyTab tab認(rèn)證文件位置 * @param principal 認(rèn)證用戶 */ public static void configureJAAS(String keyTab, String principal) { String JAAS_TEMPLATE = "KafkaClient {\n" + "com.sun.security.auth.module.Krb5LoginModule required\n" + "useKeyTab=true\n" + "keyTab=\"%1$s\"\n" + "principal=\"%2$s\";\n" + "};"; String content = String.format(JAAS_TEMPLATE, keyTab, principal); File jaasConf = null; PrintWriter writer = null; try { jaasConf = File.createTempFile("jaas", ".conf"); writer = new PrintWriter(jaasConf); writer.println(content); } catch (IOException e) { e.printStackTrace(); } finally { if (writer != null) { writer.close(); } jaasConf.deleteOnExit(); } System.setProperty("java.security.auth.login.config", jaasConf.getAbsolutePath()); } }
感謝各位的閱讀,以上就是“kafka javaAPI入庫程序的實(shí)現(xiàn)方法”的內(nèi)容了,經(jīng)過本文的學(xué)習(xí)后,相信大家對kafka javaAPI入庫程序的實(shí)現(xiàn)方法這一問題有了更深刻的體會,具體使用情況還需要大家實(shí)踐驗證。這里是創(chuàng)新互聯(lián),小編將為大家推送更多相關(guān)知識點(diǎn)的文章,歡迎關(guān)注!
當(dāng)前文章:kafkajavaAPI入庫程序的實(shí)現(xiàn)方法
網(wǎng)頁鏈接:http://bm7419.com/article22/jdeijc.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供網(wǎng)站策劃、定制開發(fā)、網(wǎng)站設(shè)計公司、電子商務(wù)、網(wǎng)站設(shè)計、外貿(mào)建站
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)