连接mongodb,kafka异步处理代码
生活随笔
收集整理的這篇文章主要介紹了
连接mongodb,kafka异步处理代码
小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.
1. mongodb異步處理
依賴:
<dependencies><dependency><groupId>org.mongodb</groupId><artifactId>mongodb-driver-async</artifactId><version>3.0.4</version></dependency></dependencies>代碼
public static void main(String[] args) {List<ServerAddress> address=new ArrayList<>();address.add(new ServerAddress("172.16.4.90",3000));address.add(new ServerAddress("172.16.4.91",3000));address.add(new ServerAddress("172.16.4.92",3000));ClusterSettings clusterSettings = ClusterSettings.builder().hosts(address).build();MongoClientSettings settings = MongoClientSettings.builder().clusterSettings(clusterSettings).build();MongoClient mongoClient = MongoClients.create(settings);MongoDatabase database = mongoClient.getDatabase("shardb");MongoCollection<Document> collection = database.getCollection("shardtable");Document doc = new Document("name", "MongoDB").append("type", "database").append("count", 1).append("info", new Document("x", 203).append("y", 102));Long start=System.currentTimeMillis();collection.insertOne(doc, new SingleResultCallback<Void>() {@Overridepublic void onResult(final Void result, final Throwable t) {System.out.println("Inserted cosume="+(System.currentTimeMillis()-start));}});System.out.println("response cosume="+(System.currentTimeMillis()-start));}?
?
2.kafka異步處理
依賴:
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>0.9.0.0</version></dependency>?
代碼
public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "172.16.4.93:9092,172.16.4.94:9092,172.16.4.95:9092");props.put("acks", "all");props.put("retries", 0);props.put("batch.size", 16384);props.put("linger.ms", 1);props.put("buffer.memory", 33554432);props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");Producer<String, String> producer = new KafkaProducer(props);Long start=System.currentTimeMillis();for(int i = 0; i < 100; i++){//Future<RecordMetadata> response= producer.send(new ProducerRecord<String, String>("davidwang456", Integer.toString(i), Integer.toString(i)),new Callback() {public void onCompletion(RecordMetadata metadata, Exception e) {if(e != null){e.printStackTrace();System.out.println("The offset of the record we just sent is: " + metadata.offset());}}}); /* if(response.isDone()){System.out.println("send message to david1 message key="+i+",value="+i);} */ }System.out.println(System.currentTimeMillis()-start);producer.close();}結(jié)果
1. kafka的異步處理結(jié)果可以打印出來。
2. mongodb的異步處理結(jié)果沒有打印出來。
轉(zhuǎn)載于:https://www.cnblogs.com/davidwang456/p/5093430.html
總結(jié)
以上是生活随笔為你收集整理的连接mongodb,kafka异步处理代码的全部?jī)?nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: jedis连接sentinel示例程序
- 下一篇: Redis的Java客户端Jedis的八