Author Archives: neohope

1、数据准备

db.person.insert({"name":"neo","age":"26","sex":"male"})
db.person.insert({"name":"joe","age":"28","sex":"male"})

2、使用Codec

class Person  
{
       public ObjectId _id;
       public double Age;  
       public String Name;  
       public String Sex;  
       
       public Person(ObjectId _id, String Name, double Age, String Sex)
       {
    	   this._id=_id;
    	   this.Name=Name;
    	   this.Age=Age;
    	   this.Sex=Sex;
       }
}

class PersonCodec implements Codec<Person> 
{
    private final CodecRegistry codecRegistry;

    public PersonCodec(final CodecRegistry codecRegistry) {
        this.codecRegistry = codecRegistry;
    }
    
    @Override
    public void encode(BsonWriter writer, Person t, EncoderContext ec) {
    	 writer.writeStartDocument();
         writer.writeName("_id");
         writer.writeObjectId(t._id);
         writer.writeName("name");
         writer.writeString(t.Name);
         writer.writeName("age");
         writer.writeDouble(t.Age);
         writer.writeName("sex");
         writer.writeString(t.Sex);
         writer.writeEndDocument();
    }

    @Override
    public Class<Person> getEncoderClass() {
        return Person.class;
    }

    @Override
    public Person decode(BsonReader reader, DecoderContext dc) 
    {
        reader.readStartDocument();
        reader.readName();
        ObjectId _id = reader.readObjectId();
        reader.readName();
        String name = reader.readString();
        reader.readName();
        double age = reader.readDouble();
        reader.readName();
        String sex =reader.readString();
        reader.readEndDocument();
        return new Person(_id,name,age,sex);
    }
}

class PersonCodecProvider implements CodecProvider 
{
    @Override
    public <T> Codec<T> get(Class<T> type, CodecRegistry cr) 
    {
        if (type == Person.class) 
        {
            return (Codec<T>) new PersonCodec(cr);
        }
        return null;
    }
}

public class CodecTest 
{
	private static void testCodec()
	{
		String[] hosts = {"127.0.0.1"};
		int port = 27017;
		String user = null;
		String password = null;
		String database = "test";
		CodecRegistry codecRegistry = CodecRegistries.fromRegistries(
	            CodecRegistries.fromProviders(new PersonCodecProvider()),
	            MongoClient.getDefaultCodecRegistry());  
		
		MongoClient mongoClient = getConnection(hosts,port,user,password,database,codecRegistry);
		MongoDatabase db = mongoClient.getDatabase("test");
		MongoCollection<Person> collection = db.getCollection("person",Person.class);
		FindIterable<Person> iterable = collection.find();
		MongoCursor<Person> cursor = iterable.iterator();
		while (cursor.hasNext())
		{
        		Person p  = cursor.next();
        		System.out.println("personName: " + p.Name);
		}
	}
	
	private static MongoClient getConnection(String[] hosts, int port, String user, String password, String database, CodecRegistry codecRegistry)
	{
	        MongoClientOptions mongoClientOptions = new MongoClientOptions.Builder()
	        .connectionsPerHost(100)
	        .threadsAllowedToBlockForConnectionMultiplier(5)
	        .maxWaitTime(1000 * 60 * 2)
	        .connectTimeout(1000 * 10)
	        .socketTimeout(0)
	        .socketKeepAlive(false)
	        .readPreference(ReadPreference.primary())
	        .writeConcern(WriteConcern.ACKNOWLEDGED)
	        .codecRegistry(codecRegistry)
	        .build();
		
		List<ServerAddress> mongoAddresses = new ArrayList<ServerAddress>();
		for (String host : hosts) {
		    mongoAddresses.add(new ServerAddress(host, port));
		}
		
		List<MongoCredential> mongoCredentials = null;
		if (user != null && !user.isEmpty() && password != null && !password.isEmpty()) {
		    mongoCredentials = new ArrayList<MongoCredential>();
		    mongoCredentials.add(MongoCredential.createMongoCRCredential(user, database, password.toCharArray()));
		}
		
		if(mongoCredentials==null)
		{
			return new MongoClient(mongoAddresses, mongoClientOptions);
		}
		else
		{
			return new MongoClient(mongoAddresses, mongoCredentials, mongoClientOptions);
		}
	}
}

1、数据准备

db.sell.insert({"price":8.0,"amount":500.0,"status":"a"})
db.sell.insert({"price":8.0,"amount":450.0,"status":"a"})
db.sell.insert({"price":8.0,"amount":400.0,"status":"a"})
db.sell.insert({"price":9.0,"amount":350.0,"status":"a"})
db.sell.insert({"price":9.0,"amount":300.0,"status":"a"})
db.sell.insert({"price":9.0,"amount":250.0,"status":"a"})
db.sell.insert({"price":9.0,"amount":200.0,"status":"a"})
db.sell.insert({"price":10.0,"amount":150.0,"status":"d"})
db.sell.insert({"price":10.0,"amount":100.0,"status":"d"})
db.sell.insert({"price":10.0,"amount":50.0,"status":"d"})
db.sell.insert({"price":10.0,"amount":0.0,"status":"d"})

2、MapReduce

	private static void testMapReduce3x()
	{
		MongoClient mongoClient = new MongoClient("localhost", 27017);
		MongoDatabase db = mongoClient.getDatabase("test");
		MongoCollection collection = db.getCollection("sell");

		String map = "function(){emit(this.price,this.amount);}";
		String reduce = "function(key, values){return Array.sum(values)}";

		MapReduceIterable out = collection.mapReduce(map, reduce);
		MongoCursor cursor = out.iterator();
		while (cursor.hasNext()) 
		{
			System.out.println(cursor.next());
		}
	}
	
	private static void testMapReduce2x()
	{
		MongoClient mongoClient = new MongoClient("localhost", 27017);
		MongoDatabase db = mongoClient.getDatabase("test");
		BasicDBObject query=new BasicDBObject("status","a");
		DBCollection dbcollection = mongoClient.getDB("test").getCollection("sell");
		
		String map = "function(){emit(this.price,this.amount);}";
		String reduce = "function(key, values){return Array.sum(values)}";
		
		MapReduceCommand cmd = new MapReduceCommand(dbcollection, map, reduce,
		    "outputCollection", MapReduceCommand.OutputType.INLINE, query);
		
		MapReduceOutput out2 = dbcollection.mapReduce(cmd);
		for (DBObject o : out2.results()) 
		{
		   System.out.println(o.toString());
		}
	}