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();}

結果

1. kafka的異步處理結果可以打印出來。

2. mongodb的異步處理結果沒有打印出來。

轉載于:https://www.cnblogs.com/davidwang456/p/5093430.html

總結

以上是生活随笔為你收集整理的连接mongodb，kafka异步处理代码的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。