storm操作mongodb之插入操作，基于storm v2.0.0版本

灰太狼 2021-12-15 04:53 197阅读 0赞

## 1、功能说明 ##

通过storm官方控件，storm-mongodb实现数据的存储，更新等相关操作。

[官网文档地址][Link 1]

## 2、依赖pom ##

<dependency>
          <groupId>org.apache.storm</groupId>
          <artifactId>storm-mongodb</artifactId>
          <version>2.0.0</version>
       </dependency>

## 3、插入文档操作 ##

插入文档经历两个过程，一实现MongoMapper，二将MongoMapper传入操作类MongoInsertBolt.bolt

String url = "mongodb://127.0.0.1:27017/test";
    String collectionName = "wordcount";
    
    MongoMapper mapper = new SimpleMongoMapper()
           .withFields("word", "count");
    
    MongoInsertBolt insertBolt = new MongoInsertBolt(url, collectionName, mapper)
       .withBatchSize(1000) //选传   批量操作
       .withFlushIntervalSecs(10) //选传 多长时间刷新一次数据到数据库
       .withOrdered(false)； //是否有序  mongoClient.insert(docs, ordered);

**接下来对MongoMapper进行详细说明**插入文档有两张实现方式，**操作一**实现*org.apache.storm.mongodb.common.mapper.MongoMapper***MongoMapper.class**接口，实现其方法

public interface MongoMapper extends Serializable {
        //需要写入mongodb的文档实体
        Document toDocument(ITuple tuple);
        //自定义生成 mongodb 主键 _id 策略 
        Document toDocumentByKeys(List<Object> keys);
    }

**操作方式二**，通过直接调用**SimpleMongoMapper.class**方法，或者继承SimpleMongoMapper.class，并实现toDocument方法，自定义输出结构。  
org.apache.storm.mongodb.common.mapper.SimpleMongoMapper

public class SimpleMongoMapper implements MongoMapper {
    
        private String[] fields;
    
        public SimpleMongoMapper(String... fields) {
            this.fields = fields;
        }
    
         /**可以通过继承SimpleMongoMapper类单独重写此方法，对tuple中的key重新定义
       *  例如：
       *     document.append("userName", tuple.getValueByField("name"));
       */
        @Override
        public Document toDocument(ITuple tuple) {
            Document document = new Document();
            for (String field : fields) {
                document.append(field, tuple.getValueByField(field));
            }
            return document;
        }
    
        @Override
        public Document toDocumentByKeys(List<Object> keys) {
            Document document = new Document();
            document.append("_id", MongoUtils.getId(keys));
            return document;
        }
    
        public SimpleMongoMapper withFields(String... fields) {
            this.fields = fields;
            return this;
        }
    }

[Link 1]: https://github.com/apache/storm/tree/v2.0.0/external/storm-mongodb