Java 类com.mongodb.WriteConcern 实例源码

/**
 * New mongo db client options factory bean.
 *
 * @param mongo the mongo properties.
 * @return the mongo client options factory bean
 */
public static MongoClientOptionsFactoryBean newMongoDbClientOptionsFactoryBean(final AbstractMongoInstanceProperties mongo) {
    try {
        final MongoClientOptionsFactoryBean bean = new MongoClientOptionsFactoryBean();
        bean.setWriteConcern(WriteConcern.valueOf(mongo.getWriteConcern()));
        bean.setHeartbeatConnectTimeout(Long.valueOf(mongo.getTimeout()).intValue());
        bean.setHeartbeatSocketTimeout(Long.valueOf(mongo.getTimeout()).intValue());
        bean.setMaxConnectionLifeTime(mongo.getConns().getLifetime());
        bean.setSocketKeepAlive(mongo.isSocketKeepAlive());
        bean.setMaxConnectionIdleTime(Long.valueOf(mongo.getIdleTimeout()).intValue());
        bean.setConnectionsPerHost(mongo.getConns().getPerHost());
        bean.setSocketTimeout(Long.valueOf(mongo.getTimeout()).intValue());
        bean.setConnectTimeout(Long.valueOf(mongo.getTimeout()).intValue());
        bean.afterPropertiesSet();
        return bean;
    } catch (final Exception e) {
        throw new BeanCreationException(e.getMessage(), e);
    }
}

public Object insert(DBCollection collection, WriteConcern writeConcern) {
    DBObject document = new BasicDBObject();
    // 匹配_id
    for (int i = 0, n = columns.size(); i < n; i++) {
        // document.put(columns.get(i), values.get(i).getValue());

        String tempColumn = columns.get(i);
        if (3 == tempColumn.length() && tempColumn.equals("_id")) {
            document.put(tempColumn, new ObjectId(values.get(i).getValue().toString()));
        } else {
            document.put(tempColumn, values.get(i).getValue());
        }
    }
    log(document);
    // TODO: WriteConcern.ACKNOWLEDGED需要可以配置
    // WriteResult result = collection.insert(document, WriteConcern.ACKNOWLEDGED);
    // collection.insert(document, MongoComponent.getInstance().getDefaultWriteConcern());
    collection.insert(document, writeConcern);
    Object oid = document.get("_id");
    if (null != oid) {
        return oid.toString();
    }
    return null;
}

/**
 * Prepares {@link WriteConcern} from given {@link WriteConcernEnum}
 *
 * @param writeConcernEnum  write concern provided in config.
 * @return  write concern for mongo client option.
 */
public static WriteConcern writeConcern(WriteConcernEnum writeConcernEnum) {
    switch (writeConcernEnum) {
        case MAJORITY:
            return WriteConcern.MAJORITY;

        case JOURNALED:
            return WriteConcern.JOURNALED;

        case ACKNOWLEDGED:
            return WriteConcern.ACKNOWLEDGED;

        case UNACKNOWLEDGED:
            return WriteConcern.UNACKNOWLEDGED;

        default:
            return null;
    }
}

public static void main(String[] args) throws IOException {
    MongoClient mongoClient = new MongoClient("localhost", 27017);
    DB db = mongoClient.getDB("mydb");
    DBCollection coll = db.getCollection("questionsCollection");
    mongoClient.setWriteConcern(WriteConcern.JOURNALED);
    GIFTParser p = new GIFTParser();
    BasicDBObject doc = null;
    for (Question q : p.parserGIFT("Data/questionsGIFT")) {
        doc = new BasicDBObject("category", q.getCategory())
                .append("question", q.getText())
                .append("correctanswer", q.getCorrectAnswer())
                .append("wrongAnswers",q.getWrongAnswers());
        coll.insert(doc);
    }

    DBCursor cursor = coll.find();
    try {
           while(cursor.hasNext()) {
               System.out.println(cursor.next());
           }
        } finally {
           cursor.close();
        }
}

public static void removeOldPackageTrack(int p_nodeid)
{
    try {

        logger.info("Purging Package Logs for NodeID: " + p_nodeid);
        DB db;
        String dbName = com.clavain.muninmxcd.p.getProperty("mongo.dbessentials");
        db = m.getDB(dbName);
        db.setWriteConcern(WriteConcern.SAFE);
        DBCollection col = db.getCollection("trackpkg");   
        BasicDBObject query = new BasicDBObject();
        query.append("node", p_nodeid);
        col.remove(query);
        db.setWriteConcern(WriteConcern.NONE);
    } catch (Exception ex)
    {
        logger.error("Error in removeOldPackageTrack: " + ex.getLocalizedMessage());
    }
}

public static boolean removeConfigurationByQuery(BasicDBObject query, MongoConnection mongoConnection) {

    DBCollection dbCol = mongoConnection.getDBCollection();

    if (dbCol == null) {
        throw new MongoConfigRuntimeException("jetstreamconfig collection is unknown");
    }

    try {
        if(query ==null) {
            return false;
        }

        WriteResult result = dbCol.remove(query, WriteConcern.SAFE);

        if(result.getLastError().ok()) {
            return true;
        }

    } catch (Exception err) {
        throw new MongoConfigRuntimeException(err);
    } 

    return true;
}

@Override
public Mongo mongo() throws Exception {

    List<ServerAddress> serverAddresses = hosts.stream()
            .map(this::createServerAddress)
            .collect(toList());

    LOG.info("Connecting to DB hosts: {}...", serverAddresses);

    if (serverAddresses.size() == 1) {
        // create a mongo client that connects to a single database,
        // this is NOT the same as calling the constructor with a list of ServerAddresses with only one element!
        return new MongoClient(serverAddresses.get(0), mongoCredentials());
    } else {
        // create a mongo client that connects to a replicaset
        MongoClientOptions options = MongoClientOptions.builder()
                .writeConcern(WriteConcern.ACKNOWLEDGED)
                .build();
        return new MongoClient(serverAddresses, mongoCredentials(), options);
    }
}

public void init(
    Stage.Context context,
    List<Stage.ConfigIssue> issues,
    ReadPreference readPreference,
    WriteConcern writeConcern
) {
  mongoClient = createClient(context, issues, readPreference, writeConcern);
  if (!issues.isEmpty()) {
    return;
  }

  mongoDatabase = createMongoDatabase(context, issues, readPreference, writeConcern);
  if (!issues.isEmpty()) {
    return;
  }

  mongoCollection = createMongoCollection(context, issues, readPreference, writeConcern);
}

private MongoDatabase createMongoDatabase(
    Stage.Context context,
    List<Stage.ConfigIssue> issues,
    ReadPreference readPreference,
    WriteConcern writeConcern
) {
  MongoDatabase mongoDatabase = null;
  try {
    if (readPreference != null) {
      mongoDatabase = mongoClient.getDatabase(database).withReadPreference(readPreference);
    } else if (writeConcern != null) {
      mongoDatabase = mongoClient.getDatabase(database).withWriteConcern(writeConcern);
    }
  } catch (MongoClientException e) {
    issues.add(context.createConfigIssue(
        Groups.MONGODB.name(),
        MONGO_CONFIG_PREFIX + "database",
        Errors.MONGODB_02,
        database,
        e.toString()
    ));
  }
  return mongoDatabase;
}

private MongoCollection createMongoCollection(
    Stage.Context context,
    List<Stage.ConfigIssue> issues,
    ReadPreference readPreference,
    WriteConcern writeConcern
) {
  MongoCollection mongoCollection = null;
  try {
    if (readPreference != null) {
      mongoCollection = mongoDatabase.getCollection(collection).withReadPreference(readPreference);
    } else if (writeConcern != null) {
      mongoCollection = mongoDatabase.getCollection(collection).withWriteConcern(writeConcern);
    }
  } catch (MongoClientException e) {
    issues.add(context.createConfigIssue(
        Groups.MONGODB.name(),
        MONGO_CONFIG_PREFIX + "collection",
        Errors.MONGODB_03,
        collection,
        e.toString()
    ));
  }
  return mongoCollection;
}

@SuppressWarnings("deprecation")
private void init()
{
    if(mongoClient != null) return;
    try {
        MongoCredential credential = MongoCredential.createCredential(
                MongoDBDrive.getInstance().getUsername(),
                MongoDBDrive.getInstance().getDatabase(),
                MongoDBDrive.getInstance().getPassword().toCharArray());
        MongoDBDrive.getInstance().mongoClient = new MongoClient(
                new ServerAddress(MongoDBDrive.getInstance().getUrl()),
                Arrays.asList(credential));
        MongoDBDrive.getInstance().mongoClient.setWriteConcern(WriteConcern.NORMAL);
    } catch (Exception e) {
        return;
    }
    return;
}

public static void connect(String server, int port, String dbname, String user, String pass) throws UnknownHostException {
    disconnect();

    ServerAddress sa = new ServerAddress(server, port);

    if (user != null && user.length() > 0) {
        List<MongoCredential> creds = new ArrayList<>();
        creds.add(MongoCredential.createScramSha1Credential(user, dbname, pass.toCharArray()));
        Connection = new MongoClient(sa, creds);
    } else {
        Connection = new MongoClient(sa);
    }

    Database = Connection.getDatabase(dbname);
    Server = server;
    Port = port;
    DatabaseName = dbname;

    User = user;
    Password = pass;

    Connection.setWriteConcern(WriteConcern.SAFE);
    Database.listCollectionNames().first();
}

@Test
public void testUpload() throws IOException {

    MongoFileStoreConfig config = MongoFileStoreConfig.builder()//
            .bucket("mongofs").chunkSize(ChunkSize.medium_256K)//
            .enableCompression(true).enableEncryption(new BasicCrypto())//
            .writeConcern(WriteConcern.SAFE) //
            .build();
    MongoFileStore store = new MongoFileStore(database, config);

    ByteArrayInputStream in = new ByteArrayInputStream(LoremIpsum.LOREM_IPSUM.getBytes());
    MongoFile mongoFile = store.upload("loremIpsum.txt", "test/plain", null, false, in);
    assertNotNull(mongoFile);

    assertEquals(32087, mongoFile.getLength());

}

@Override
public void setReadWriteMode(String readWriteMode) {
    if (readWriteMode == null || readWriteMode.equals(lastReadWriteMode)) {
        return;
    }
    lastReadWriteMode = readWriteMode;
    try {
        Map<String, String> map = Splitter.on(", ").withKeyValueSeparator(":").split(readWriteMode);
        String read = map.get("read");
        if (read != null) {
            ReadPreference readPref = ReadPreference.valueOf(read);
            if (!readPref.equals(this.readPreference)) {
                this.readPreference = readPref;
            }
        }
        String write = map.get("write");
        if (write != null) {
            WriteConcern writeConcern = WriteConcern.valueOf(write);
            if (!writeConcern.equals(this.writeConcern)) {
                this.writeConcern = writeConcern;
            }
        }
    } catch (Exception e) {
        // unsupported or parse error - ignore
    }
}

@Nonnull
@Override
public Entry newEntry(@Nonnull final Revision from,
                      @Nonnull final Revision to) {
    return new MemoryEntry(from, to) {

        private Diff commit = new Diff(from, to);

        @Override
        public void append(@Nonnull String path, @Nonnull String changes) {
            // super.append() will apply to diff cache in base class
            super.append(path, changes);
            commit.append(path, changes);
        }

        @Override
        public void done() {
            try {
                changes.insert(commit.doc, WriteConcern.UNACKNOWLEDGED);
            } catch (MongoException e) {
                LOG.warn("Write back of diff cache entry failed", e);
            }
        }
    };
}

@Ignore
@Test
public void performBenchMark_WriteConcern() throws UnknownHostException, InterruptedException {
    Mongo mongo = new Mongo(new DBAddress(remoteServer));
    final DB db = mongo.getDB(TEST_DB1);
    final DBCollection nodes = db.getCollection("nodes");
    final DBCollection blobs = db.getCollection("blobs");
    int readers = 0;
    int writers = 2;
    for(WriteConcern wc : namedConcerns.keySet()){
        prepareDB(nodes,blobs);
        final Benchmark b = new Benchmark(nodes, blobs);
        Result r = b.run(readers, writers, true, wc);
        results.add(r);
    }

    prepareDB(nodes,blobs);

    dumpResult();
}

private MongoDB() {
   MongoClientOptions mongoOptions = MongoClientOptions.builder()
.socketTimeout(60000) // Wait 1m for a query to finish, https://jira.mongodb.org/browse/JAVA-1076
.connectTimeout(15000) // Try the initial connection for 15s, http://blog.mongolab.com/2013/10/do-you-want-a-timeout/
.maxConnectionIdleTime(600000) // Keep idle connections for 10m, so we discard failed connections quickly
.readPreference(ReadPreference.primaryPreferred()) // Read from the primary, if not available use a secondary
.build();
   MongoClient mongoClient;
   mongoClient = new MongoClient(new ServerAddress(DB_HOST, DB_PORT), mongoOptions);

   mongoClient.setWriteConcern(WriteConcern.SAFE);
   datastore = new Morphia().mapPackage(BaseEntity.class.getPackage().getName())
.createDatastore(mongoClient, DB_NAME);
   datastore.ensureIndexes();
   datastore.ensureCaps();
   LOG.info("Connection to database '" + DB_HOST + ":" + DB_PORT + "/" + DB_NAME + "' initialized");
 }

public void deleteExternalData(
        DBCollection snapsIdx, 
        DBCollection snapsHtml, 
        DBCollection snapsImag, 
        DBCollection snapsThumb, 
        WriteConcern writeConcern) {

    // delete all inverse index data
    Idx.removeForSnap(snapsIdx, getId(), writeConcern);

    // remove html
    SnapHtml.deleteForSnap(snapsHtml, getId(), writeConcern);

    // remove images
    SnapImag.deleteForSnap(snapsImag, getId(), writeConcern);
    SnapThumb.deleteForSnap(snapsThumb, getId(), writeConcern);
}

public final static boolean upsert(DBCollection coll, DBObject q, DBObject dbo, boolean ensureId) {

        WriteResult wr = coll.update(q, dbo, true, false, WriteConcern.ACKNOWLEDGED);
        boolean updatedExisting = wr.isUpdateOfExisting();

        if (ensureId) {
            if (updatedExisting) {

                BasicDBObject f = new BasicDBObject();
                f.put(MongoUtils._id, 1);

                DBObject o = coll.findOne(q, f);
                dbo.put(MongoUtils._id, o.get(MongoUtils._id));

            } else {
                ObjectId upserted = (ObjectId) wr.getUpsertedId();
                dbo.put(MongoUtils._id, upserted);
            }
        }

        return updatedExisting;     
    }

@POST
@Path("park")
@Consumes("application/json")
@Produces("application/json")
public HashMap insertAPark(Park park){
    BasicDBObject parkObject = new BasicDBObject("Name",park.getName());
    parkObject.append("pos", park.getPos());

    DB db = dbConnection.getDB();
    DBCollection parkListCollection = db.getCollection("parkpoints");
    try{
        parkListCollection.insert(parkObject, WriteConcern.SAFE);
    } catch (Exception e) {
        System.out.println("threw an exception: " + e.getClass() + " :: " + e.getMessage());
    }

    //now make it look pretty
    HashMap holder = new HashMap<String, Object>();
    holder.put("name",parkObject.get("Name"));
    holder.put("position", parkObject.get("pos"));
    holder.put("id", parkObject.get("_id").toString());
    return holder;

}

/**
 * Attempts to insert a lock record to the db
 *
 * @returns true if successful, false if lock already exists. Any other case
 * would be an exception.
 */
private boolean acquire(String callerId, String resourceId, Long ttl, Date now, Date expiration) {
    BasicDBObject update = new BasicDBObject().
            append(CALLERID, callerId).
            append(RESOURCEID, resourceId).
            append(TIMESTAMP, now).
            append(TTL, ttl).
            append(EXPIRATION, expiration).
            append(COUNT, 1).
            append(VERSION, 1);

    try {
        LOGGER.debug("insert: {}", update);
        coll.insert(update, WriteConcern.ACKNOWLEDGED);
    } catch (DuplicateKeyException dke) {
        return false;
    }
    return true;
}

public void ping(String callerId, String resourceId) {
    Date now = new Date();
    BasicDBObject q = new BasicDBObject().
            append(CALLERID, callerId).
            append(RESOURCEID, resourceId).
            append(EXPIRATION, new BasicDBObject("$gt", now)).
            append(COUNT, new BasicDBObject("$gt", 0));
    DBObject lock = coll.findOne(q,null,ReadPreference.primary());
    if (lock != null) {
        Date expiration = new Date(now.getTime() + ((Number) lock.get(TTL)).longValue());
        int ver = ((Number) lock.get(VERSION)).intValue();
        BasicDBObject update = new BasicDBObject().
                append("$set", new BasicDBObject(TIMESTAMP, now).
                        append(EXPIRATION, expiration)).
                append("$inc", new BasicDBObject(VERSION, 1));
        q = q.append(VERSION, ver);
        WriteResult wr = coll.update(q, update, false, false, WriteConcern.ACKNOWLEDGED);
        if (wr.getN() != 1) {
            throw new InvalidLockException(resourceId);
        }
        LOGGER.debug("{}/{} pinged", callerId, resourceId);
    } else {
        throw new InvalidLockException(resourceId);
    }
}

@Test
public void readPreference() throws IOException {
    try (InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream("parse-test-datasources.json")) {
        JsonNode node = JsonUtils.json(is);

        MongoConfiguration metadataConfig = new MongoConfiguration();
        metadataConfig.initializeFromJson(node.get("metadata_readPreference"));

        MongoConfiguration dataConfig = new MongoConfiguration();
        dataConfig.initializeFromJson(node.get("mongodata_readPreference"));

        assertEquals(ReadPreference.nearest(), metadataConfig.getMongoClientOptions().getReadPreference());
        assertEquals(ReadPreference.secondary(), dataConfig.getMongoClientOptions().getReadPreference());
        assertEquals(WriteConcern.SAFE, metadataConfig.getWriteConcern());
    }
}

@POST
@Path("park")
@Consumes("application/json")
@Produces("application/json")
public HashMap insertAPark(Park park){
    BasicDBObject parkObject = new BasicDBObject("Name",park.getName());
    parkObject.append("pos", park.getPos());

    DB db = dbConnection.getDB();
    DBCollection parkListCollection = db.getCollection("parks");
    try{
        parkListCollection.insert(parkObject, WriteConcern.SAFE);
    } catch (Exception e) {
        System.out.println("threw an exception: " + e.getClass() + " :: " + e.getMessage());
    }

    //now make it look pretty
    HashMap holder = new HashMap<String, Object>();
    holder.put("name",parkObject.get("Name"));
    holder.put("position", parkObject.get("pos"));
    holder.put("id", parkObject.get("_id").toString());
    return holder;

}

public void init() throws Exception {
    mongoClient = new Mongo(getMongoServer());
    mongoClient.setWriteConcern(WriteConcern.SAFE);     // throw exceptions on failed write
    database = mongoClient.getDB("jagornet-dhcpv6");
    log.info("Connected to jagornet-dhcpv6 via Mongo client: " + mongoClient.toString());
    dhcpLeases = database.getCollection("DHCPLEASE");
    dhcpLeases.ensureIndex(new BasicDBObject("ipAddress", 1), "pkey", true);
    dhcpLeases.ensureIndex(new BasicDBObject("duid", 1)
                                .append("iatype", 1)
                                .append("iaid", 1),
                                "tuple", false);
    dhcpLeases.ensureIndex("duid");
    dhcpLeases.ensureIndex("iatype");
    dhcpLeases.ensureIndex("state");
    dhcpLeases.ensureIndex("validEndTime");
}

public MongoV3(List<ServerAddress> servers, List<MongoCredential> authors) {
    Builder options = new MongoClientOptions.Builder();
    options.connectionsPerHost(50);// 连接池设置为300个连接,默认为100
    options.connectTimeout(15000);// 连接超时，推荐>3000毫秒
    options.maxWaitTime(5000); //
    options.socketTimeout(500);
    options.writeConcern(WriteConcern.W2);
    con = new MongoClient(servers, authors, options.build());
    setMongoConnect(this);
}

@Override
public void start() throws IOException {
  MongoClientURI clientURI = new MongoClientURI(mongoURL);
  client = new MongoClient(clientURI);
  DB db = client.getDB(clientURI.getDatabase());
  collection = db.getCollection(clientURI.getCollection());
  collection.setWriteConcern(WriteConcern.JOURNALED);
  DBObject index = new BasicDBObject(1).append(pKey, Integer.valueOf(1));
  collection.createIndex(index);
}

@Override
public WriteConcern getDefaultWriteConcern(String dsKey) {
    AbstractMongoDataSource dataSource = realDataSourceMap.get(dsKey);
    if (null == dataSource) {
        throw new DataSourceException("A non-existent mongo data source: " + dsKey);
    }
    return dataSource.getDefaultWriteConcern();
}

public int delete(DBCollection collection, WriteConcern writeConcern) {
    DBObject query = new BasicDBObject();
    if (null != condition) {
        this.condition.setQuery(query, null);
    }

    log(query);

    // WriteResult result = collection.remove(query, WriteConcern.ACKNOWLEDGED);
    WriteResult result = collection.remove(query, writeConcern);
    // collection.remove(query)
    // System.out.println(query.toString());
    return result.getN();
}

private WriteConcern determineWriteConcern(String label) {
    switch (label.toLowerCase()) {
        case "acknowledged":
            return WriteConcern.ACKNOWLEDGED;
        case "unacknowledged":
            return WriteConcern.UNACKNOWLEDGED;
        case "journaled":
            return WriteConcern.JOURNALED;
        case "majority":
            return WriteConcern.MAJORITY;
        default:
            return WriteConcern.ACKNOWLEDGED;
    }
}

@Override
public Datastore get() {
    if (ds == null){
        MongoClient mc = new MongoClient(hostname, MongoClientOptions.builder()
                .writeConcern(WriteConcern.ACKNOWLEDGED)
                .build());
        ds = morphia.createDatastore(mc, database);
    }
    return ds;
}

public WriteConcern toWriteConcern(Object obj, WriteConcern defaultValue) {
    if(obj instanceof WriteConcern) return (WriteConcern) obj;
    if(decision.isSimpleValue(obj)) {
        String str = caster.toString(obj,"");
        str=str.trim().toUpperCase();
        if("ACKNOWLEDGED".equals(str))
            return WriteConcern.ACKNOWLEDGED;
        else if("ACKNOWLEDGED".equals(str))
            return WriteConcern.FSYNC_SAFE;
        else if("FSYNC_SAFE".equals(str) || "FSYNCSAFE".equals(str))
            return WriteConcern.FSYNCED;
        else if("JOURNAL_SAFE".equals(str) || "JOURNALSAFE".equals(str))
            return WriteConcern.JOURNAL_SAFE;
        else if("JOURNALED".equals(str))
            return WriteConcern.JOURNALED;
        else if("MAJORITY".equals(str))
            return WriteConcern.MAJORITY;
        else if("NORMAL".equals(str))
            return WriteConcern.NORMAL;
        else if("REPLICA_ACKNOWLEDGED".equals(str) || "REPLICAACKNOWLEDGED".equals(str))
            return WriteConcern.REPLICA_ACKNOWLEDGED;
        else if("REPLICAS_SAFE".equals(str) || "REPLICASSAFE".equals(str))
            return WriteConcern.REPLICAS_SAFE;
        else if("SAFE".equals(str))
            return WriteConcern.SAFE;
        else if("UNACKNOWLEDGED".equals(str))
            return WriteConcern.UNACKNOWLEDGED;
    }
    return defaultValue;
}

private static MongoClient getInstance(){
    if (client == null) {
        Properties prop = PropertiesUtil.loadProperties("mongodb.properties");
        String host = PropertiesUtil.getString(prop, "host");
        int port = PropertiesUtil.getInt(prop, "port");
        defauleDbName = "admin";
        try {
            client = new MongoClient(host, port);
        } catch (UnknownHostException e) {
            logger.info("{}", e);
        }

        // or, to connect to a replica set, with auto-discovery of the primary, supply a seed list of members
        // List<ServerAddress> listHost = Arrays.asList(new ServerAddress("localhost", 27017),new ServerAddress("localhost", 27018));
        // instance.mongoClient = new MongoClient(listHost);
        // 大部分用户使用mongodb都在安全内网下，但如果将mongodb设为安全验证模式，就需要在客户端提供用户名和密码：
        //boolean auth = client.authenticate(myUserName, myPassword);

        Builder options = new MongoClientOptions.Builder();
        options.connectionsPerHost(300);        // 连接池设置为300个连接,默认为100
        options.connectTimeout(15000);          // 连接超时，推荐>3000毫秒
        options.maxWaitTime(5000); 
        options.socketTimeout(0);               // 套接字超时时间，0无限制
        options.threadsAllowedToBlockForConnectionMultiplier(5000);// 线程队列数，如果连接线程排满了队列就会抛出 "Out of semaphores to get db"错误。
        options.writeConcern(WriteConcern.SAFE);
        options.build();
    }
    return client;
}

private void configureClientOptions(final Map<String, Object> properties) {
    final MongoClientOptions.Builder builder = MongoClientOptions.builder();
    final String writeConcern = (String) properties.get(ECLIPSELINK_NOSQL_PROPERTY_MONGO_WRITE_CONCERN);
    final String readPreference = (String) properties.get(ECLIPSELINK_NOSQL_PROPERTY_MONGO_READ_PREFERENCE);

    if (writeConcern != null) {
        builder.writeConcern(WriteConcern.valueOf(writeConcern));
    }
    if (readPreference != null) {
        builder.readPreference(ReadPreference.valueOf(readPreference));
    }

    mongoClientOptions = builder.build();
}

private void configureClientOptions(final Map<String, Object> properties) {
    final MongoClientOptions.Builder builder = MongoClientOptions.builder();
    setOptions(builder, (final String key) -> (String) properties.get(HIBERNATE_OGM_MONGODB_OPTIONS_PREFIX + "." + key));

    final String writeConcern = (String) properties.get(HIBERNATE_OGM_MONGODB_WRITE_CONCERN);
    final String readPreference = (String) properties.get(HIBERNATE_OGM_MONGODB_READ_PREFERENCE);

    if (writeConcern != null) {
        builder.writeConcern(WriteConcern.valueOf(writeConcern));
    }
    if (readPreference != null) {
        builder.readPreference(ReadPreference.valueOf(readPreference));
    }
    mongoClientOptions = builder.build();
}

@Test
public void testMongoClientOptions() {
    // GIVEN
    final Map<String, Object> properties = new HashMap<>();
    when(descriptor.getProperties()).thenReturn(properties);

    properties.put("eclipselink.nosql.property.mongo.db", "foo");

    // it looks like only the two options below are supported by EclipseLink

    final ReadPreference readPreference = ReadPreference.nearest();
    final WriteConcern writeConcern = WriteConcern.JOURNALED;

    properties.put("eclipselink.nosql.property.mongo.read-preference", readPreference.getName());
    properties.put("eclipselink.nosql.property.mongo.write-concern", "JOURNALED");

    final ConfigurationFactory factory = new ConfigurationFactoryImpl();

    // WHEN
    final Configuration configuration = factory.createConfiguration(descriptor);

    // THEN
    assertThat(configuration, notNullValue());

    final MongoClientOptions clientOptions = configuration.getClientOptions();
    assertThat(clientOptions, notNullValue());
    assertThat(clientOptions.getReadPreference(), equalTo(readPreference));
    assertThat(clientOptions.getWriteConcern(), equalTo(writeConcern));
}

public MongoTemplate getMongoTemplate(String database)
    throws UnknownHostException, MongoException {
    if (!tmptMap.containsKey(database)) {
        synchronized (this) {
            if (!tmptMap.containsKey(database)) {
                NaviMongoDbFactory dbFactory = new NaviMongoDbFactory(dataSource, database);
                MongoTemplate template = new MongoTemplate(dbFactory, getDefaultMongoConverter(dbFactory));
                template.setWriteConcern(WriteConcern.SAFE);
                tmptMap.put(database, template);
            }
        }
    }
    return tmptMap.get(database);

}

@Override
public void start() throws IOException {
  MongoClientURI clientURI = new MongoClientURI(mongoURL);
  client = new MongoClient(clientURI);
  MongoDatabase db = client.getDatabase(clientURI.getDatabase());
  collection = db.getCollection(clientURI.getCollection()).withWriteConcern(WriteConcern.JOURNALED);
  Bson index = Indexes.ascending(pKey);
  collection.createIndex(index);
}

/**
 * Set the {@link WriteConcern} for write operations on MongoDB, passing in the bean ref to a custom WriteConcern which exists in the Registry.
 * You can also use standard WriteConcerns by passing in their key. See the {@link #setWriteConcern(String) setWriteConcern} method.
 * 
 * @param writeConcernRef the name of the bean in the registry that represents the WriteConcern to use
 */
public void setWriteConcernRef(String writeConcernRef) {
    WriteConcern wc = this.getCamelContext().getRegistry().lookupByNameAndType(writeConcernRef, WriteConcern.class);
    if (wc == null) {
        String msg = "Camel MongoDB component could not find the WriteConcern in the Registry. Verify that the "
                + "provided bean name (" + writeConcernRef + ")  is correct. Aborting initialization.";
        throw new IllegalArgumentException(msg);
    }

    this.writeConcernRef = wc;
}

/**
 * Set the {@link WriteConcern} for write operations on MongoDB, passing in the bean ref to a custom WriteConcern which exists in the Registry.
 * You can also use standard WriteConcerns by passing in their key. See the {@link #setWriteConcern(String) setWriteConcern} method.
 * 
 * @param writeConcernRef the name of the bean in the registry that represents the WriteConcern to use
 */
public void setWriteConcernRef(String writeConcernRef) {
    WriteConcern wc = this.getCamelContext().getRegistry().lookupByNameAndType(writeConcernRef, WriteConcern.class);
    if (wc == null) {
        String msg = "Camel MongoDB component could not find the WriteConcern in the Registry. Verify that the "
                + "provided bean name (" + writeConcernRef + ")  is correct. Aborting initialization.";
        throw new IllegalArgumentException(msg);
    }

    this.writeConcernRef = wc;
}